Runner

PLC is the execution engine. It takes a program, holds the current state, and exposes methods to drive execution scan by scan.

Creating a runner

from pyrung import PLC

runner = PLC(logic)

The constructor accepts:

A Program (the common case)
A list of rungs ([rung1, rung2])
None for an empty program (useful in tests)

Optional keyword arguments:

initial_state — a SystemState to start from instead of the default
history — retention window for the scan log and checkpoints. Duration string ("1h", "30m"), scan count (int), or None (unlimited, default). Prevents unbounded memory growth on long runs.
cache — instant-lookup window for full SystemState snapshots. Same formats as history. None (default) uses byte-budget-only eviction.
history_budget — byte ceiling for the recent-state cache (default: 100 MB; minimum 1 MB). Acts as a safety net when duration-based policies aren't enough.

Time modes

runner = PLC(logic, dt=0.010)        # fixed-step, 10 ms per scan (default)
runner = PLC(logic, realtime=True)   # wall-clock

Mode	Behavior	Use case
`dt=0.010`	`timestamp += dt` each scan	Tests, offline simulation
`realtime=True`	`timestamp` = actual elapsed time	Live hardware, integration tests

dt= is the default. Timer and counter instructions use timestamp, so fixed steps give perfectly reproducible results. realtime=True is intentionally non-deterministic — scan dt follows host elapsed time.

Real-time clock

Logic that depends on time of day (shift changes, scheduled events) uses the RTC system points (system.rtc.year4, system.rtc.month, system.rtc.hour, etc.). By default, these track wall-clock time. See System points for the full namespace overview.

set_rtc pins the RTC to a specific datetime:

from datetime import datetime

runner.set_rtc(datetime(2026, 3, 5, 6, 59, 50))

The RTC then advances with simulation time: rtc = base_datetime + (current_sim_time - sim_time_at_set). With a fixed dt, this makes time-of-day logic fully deterministic. With realtime=True, it effectively offsets the wall clock.

Execution methods

`step()` — one scan

state = runner.step()

Executes one complete scan cycle (all phases) and returns the committed SystemState.

`run(cycles)` — N scans

state = runner.run(cycles=300)

Runs exactly N scans, unless a pause breakpoint fires first. Returns the final state.

`run_for(seconds)` — advance by time

state = runner.run_for(1.0)  # advance simulation clock by at least 1 second

Keeps stepping until the simulation clock has advanced by the given amount (or a pause breakpoint fires).

`run_until(*conditions)` — stop on condition

state = runner.run_until(~MotorRunning, max_cycles=10000)

Accepts the same condition expressions used inside rung(). Multiple conditions are AND-ed:

runner.run_until(Motor, ~Fault)
runner.run_until(Temp > 150.0)
runner.run_until(Or(AlarmA, AlarmB, AlarmC))

Stops when the condition is true, a pause breakpoint fires, or max_cycles is reached — whichever comes first.

`run_until_fn(predicate)` — callable predicate

For conditions that aren't expressible as tag/condition expressions:

state = runner.run_until_fn(
    lambda s: s.scan_id >= 100,
    max_cycles=10000,
)

The predicate receives the committed SystemState each scan.

Injecting inputs

`patch()` — one-shot

runner.patch({Button: True, Step: 5})

Values are applied at the start of the next step() and then discarded. Multiple patches before a step merge — last write per tag wins.

`.value` via context manager

Inside with PLC(...) as plc: (or with runner:), tag .value reads and writes go through the runner's current state:

with PLC(logic) as plc:
    Button.value = True       # queues a patch
    print(Step.value)         # reads current value
    plc.step()                # executes with the queued patch
    assert Motor.value is True

Forces

Forces persist across scans, re-applied at two points each scan:

Phase 3: APPLY FORCES (pre-logic)    ← sets force values before any rung runs
Phase 4: EXECUTE LOGIC               ← logic may overwrite forced values mid-scan
Phase 5: APPLY FORCES (post-logic)   ← re-asserts force values after all logic

This means:

Forced values are present at scan start and scan end.
Logic may temporarily change a forced value mid-scan (for example, latch() on a forced-False tag sets it True temporarily, but the post-logic force pass restores it).
Edge detection (rise/fall) sees the post-force values that carry across scans.

If a tag is both patched and forced in the same scan, the pre-logic force pass overwrites the patched value. The patch is consumed but has no effect.

For force usage patterns in tests, see Testing — Forces.

Mode control

`stop()` — enter STOP mode

runner.stop()

Sets PLC mode to STOP. Does not clear tags. Idempotent.

Auto-restart from STOP

Any execution method (step, run, run_for, run_until) performs a STOP→RUN transition before executing:

Non-retentive tags reset to defaults
Retentive tags preserve values
Runtime scope resets (scan_id=0, timestamp=0.0, history/patches/forces cleared)

`reboot()` — power-cycle

runner.reboot()

Simulates a power cycle. Tag behavior depends on battery:

Battery present (default): all tags preserve
Battery absent: all tags reset to defaults

Runtime scope resets the same as STOP→RUN. Runner returns in RUN mode.

runner.battery_present = False
runner.reboot()  # all tags reset

Inspecting state

runner.state            # SystemState snapshot at latest committed scan
runner.simulation_time  # shorthand for state.timestamp
runner.dt               # configured scan period in seconds
runner.time_mode        # current TimeMode
runner.forces           # read-only view of active force overrides

SystemState fields:

state.scan_id    # int — monotonic scan counter (starts at 0)
state.timestamp  # float — simulation clock in seconds
state.tags       # PMap[str, value] — all tag values
state.memory     # PMap[str, value] — internal engine state

Both scan_id and timestamp reset to 0 on STOP→RUN transition or reboot().

History

Enable history retention to keep immutable state snapshots:

runner = PLC(logic)  # 100 MB default cache, all scans addressable

runner.history.at(5)          # state at scan 5
runner.history.range(3, 7)    # [scan 3, 4, 5, 6]

transition = runner.history.previous_transition(
    Step,
    to=101,
    at_or_before=runner.playhead,
)
runner.history.latest(10)     # up to 10 most recent (oldest → newest)

Every scan from 0 to the current tip is addressable. Recent scans are served from an in-memory state cache (byte-bounded, default 100 MB); older scans are reconstructed on demand from the scan log and checkpoints.

previous_transition() returns the latest matching recorded transition, or None. It uses compressed history indexes where available, independent of whether the underlying history is encoded as stable, alternating, arithmetic, or opaque ranges.

To bound memory on long runs, set a retention window:

runner = PLC(logic, history="1h")                   # keep 1 hour of replayable history
runner = PLC(logic, history="1h", cache="5m")       # last 5 minutes instant, rest via replay
runner = PLC(logic, history_budget=20 * 1024 * 1024)  # 20 MB byte ceiling

history_budget must be at least 1 MB (raises ValueError below that).

Time-travel playhead

The playhead is a read-only cursor into history. It doesn't affect execution — step() always appends at the history tip.

runner.playhead              # current inspection scan_id
runner.seek(scan_id=5)       # jump to a historical scan
runner.rewind(seconds=1.0)   # move backward by simulation time

snapshot = runner.history.at(runner.playhead)

rewind(seconds) finds the nearest state where timestamp <= target.

Diff

Compare two retained scans to see what changed:

changes = runner.diff(scan_a=5, scan_b=10)
# {"Motor": (True, False), "Step": (3, 7)}

Returns string-keyed dicts — only tags whose values differ. Missing tags appear as None.

Fork

Create an independent runner from a snapshot:

alt = runner.fork()              # from current state (common case)
alt = runner.fork(scan_id=10)    # from a retained historical scan
alt = runner.fork_from(scan_id=10)  # alias

The fork starts with the snapshot's state and the same time mode. It has clean runtime state — no forces, patches, breakpoints, or monitors carry over. Only the fork snapshot is in its initial history.

See Testing — Forking for the alternate-outcomes pattern.

Breakpoints and monitors

when() creates condition breakpoints evaluated after each committed scan. monitor() watches a tag for value changes. Both return handles with .remove(), .enable(), .disable().

runner.when(Fault).pause()                   # halt run()/run_for()/run_until()
runner.when(Fault).snapshot("fault_seen")    # label scan in history
runner.when(~Sensor).do(lambda state: runner.patch({Sensor: True}))  # run a callback, keep going
runner.monitor(Motor, lambda curr, prev: print(f"{prev} → {curr}"))

.do() runs a callback every scan its condition holds and lets the run continue — the hook for reactive inputs. Paired with patch, it re-asserts an input whenever the program drifts it: when(~Sensor).do(...) patches Sensor back True each time it falls (use patch, not force — a force would pin the tag, so the program could never drift it and the reaction would be moot). The patch changes visible state, so run_until(..., fold=True) won't fold past it — the window steps scan-by-scan while the callback is active, then folds normally once it stops firing.

See Testing — Monitoring changes and Testing — Predicate breakpoints for usage patterns.