Skip to content

Runtime API

Tier: Stable Core

Runner lifecycle, system points, timebase helpers, and feedback harness support.

pyrung.PLC

Generator-driven PLC execution engine.

Executes PLC logic as pure functions: Logic(state) -> new_state. The consumer controls execution via step(), enabling: - Input injection via patch() - Inspection of retained historical state via runner.history - Pause/resume at any scan boundary

Attributes:

Name Type Description
current_state SystemState

The current SystemState snapshot.

history History

Query interface for retained SystemState snapshots.

simulation_time float

Current simulation clock (seconds).

dt float

Configured scan period in seconds.

time_mode TimeMode

Current time mode (FIXED_STEP or REALTIME).

program property

program: Any

The Program object if the PLC was constructed from one, else None.

state property

state: SystemState

Current state snapshot.

tags property

tags: MappingProxyType[str, Tag]

Read-only mapping of tag name → Tag object.

bounds_violations property

bounds_violations: dict[str, BoundsViolation]

Constraint violations from the most recent scan, if any.

history property

history: History

Read-only history query surface.

playhead property

playhead: int

Current scan id used for inspection/time-travel queries.

dataview property

dataview: Any

Chainable query over this program's tag dependency graph.

Convenience shorthand for plc.program.dataview() — builds (and caches) the static program graph lazily on first access.

query property

query: Any

Survey namespace for whole-program dynamic analysis.

simulation_time property

simulation_time: float

Current simulation clock in seconds.

dt property

dt: float

Configured scan period in seconds.

time_mode property

time_mode: TimeMode

Current time mode.

debug property

debug: _DebugNamespace

Debugger-facing methods and internal runtime access.

battery_present property writable

battery_present: bool

Simulated backup battery presence.

forces property

forces: Mapping[str, bool | int | float | str]

Read-only view of active persistent overrides.

seek

seek(scan_id: int) -> SystemState

Move playhead to a retained scan and return that snapshot.

rewind

rewind(seconds: float) -> SystemState

Move playhead backward in time by seconds and return snapshot.

rung_firings

rung_firings(scan_id: int | None = None) -> PMap

Return rung firings for the given scan (default: playhead).

Returns PMap[int, PMap[str, Any]] mapping each rung index that fired (had any write, even if all were filtered by PDG) during the scan to the filtered {tag_name: value_written} map. Synthesized on demand from per-rung range-encoded timelines (:class:RungFiringTimelines); rungs with no timeline covering scan_id contribute nothing.

Populated uniformly by both the non-debug (step() / run()) and debug (DAP pyrungStepScan / continue) scan paths via ScanContext.capturing_rung.

Values are attempted writes, including a later writer reasserting the already-pending value. Transition consumers compare committed state; causal execution consumers retain the exact writer occurrence.

diff

diff(
    scan_a: int, scan_b: int
) -> dict[str, tuple[Any, Any]]

Return changed tag values between two retained historical scans.

cause

cause(
    tag: Tag | str,
    scan: int | None = None,
    *,
    to: Any = _SENTINEL,
    assume: dict[str, Any] | None = None,
    deep: bool = True,
    since: int | None = None,
) -> CausalChain | None

Explain what caused a tag to transition.

Recorded (default, to omitted): walks recorded history backward from the transition. Returns None if no transition was found. With deep=True, each fired rung's steady enablers are recursively explained through their establishing transitions or an observed writer of their current value. Non-firing static writers are not swept into the result. Terminals are classified in :attr:~pyrung.core.analysis.causal.CausalChain.roots.

Projected (to=value): projects forward from the current state, finding reachable paths that would drive the tag to value. Returns a CausalChain with mode='projected' (reachable) or mode='unreachable' (stranded, with blockers). Never returns None in projected mode.

Parameters:

Name Type Description Default
tag Tag | str

Tag object or tag name string.

required
scan int | None

Specific scan to examine (recorded mode only).

None
to Any

Target value for projected mode. When provided, the method returns the path that would drive tag to this value from the current state.

_SENTINEL
assume dict[str, Any] | None

Tag-to-value overrides for projected mode. Pins the given tags to specified values during analysis. Raises ValueError if used without to= or if any key is a readonly tag.

None
deep bool

Recorded mode only. False restores the shallow trigger-only walk (no recursive enabler explanation).

True
since int | None

Recorded mode only. Inclusive earliest transition scan for an incident-local explanation. The state immediately before this scan is boundary context, not a cause to chase. Omit it to traverse the runner's complete inherited epoch lineage.

None

Returns:

Name Type Description
A CausalChain | None

class:~pyrung.core.analysis.causal.CausalChain, or None

CausalChain | None

(recorded mode only, when no transition was found).

effect

effect(
    tag: Tag | str,
    scan: int | None = None,
    *,
    from_: Any = _SENTINEL,
    to_value: Any = _SENTINEL,
    assume: dict[str, Any] | None = None,
    steady_state_k: int = 3,
    max_scans: int = 1000,
) -> CausalChain | None

Trace the downstream effects of a tag transition.

Recorded (default, from_ omitted): walks recorded history forward from an actual transition. Returns None if no transition was found.

Projected (from_=value): what-if analysis — if the tag transitioned from value right now, what downstream effects would follow? Returns mode='projected' (possibly empty steps for dead-end) or mode='unreachable' if the trigger can't fire. Never returns None in projected mode.

Parameters:

Name Type Description Default
tag Tag | str

Tag object or tag name string.

required
scan int | None

Specific scan of the transition (recorded mode only).

None
from_ Any

Current value for projected what-if analysis. For Bool tags the TO value is inferred as not from_.

_SENTINEL
assume dict[str, Any] | None

Tag-to-value overrides for projected mode. Pins the given tags to specified values during analysis. Raises ValueError if used without from_= or if any key is a readonly tag.

None
steady_state_k int

Stop after this many consecutive scans with no new effects (recorded mode only, default 3).

3
max_scans int

Hard cap on forward scans (recorded mode only, default 1000).

1000

Returns:

Name Type Description
A CausalChain | None

class:~pyrung.core.analysis.causal.CausalChain, or None

CausalChain | None

(recorded mode only, when no transition was found).

why

why(*tags: Tag | str) -> CausalChain

Explain how tags reached their current values from a snapshot.

No history required. Walks the program graph backward from each tag, using the current state as evidence. Terminates at external inputs.

Multiple tags produce one unified tree — a single explanation consistent with all observations.

Parameters:

Name Type Description Default
tags Tag | str

One or more Tag objects or tag name strings.

()

Returns:

Name Type Description
A CausalChain

class:~pyrung.core.analysis.causal.CausalChain with

CausalChain

mode='why'.

how

how(
    *conditions: Any,
    avoid: Any = None,
    max_scans: int = 4000,
    unlink: list[str] | None = None,
    on_event: Any = None,
) -> Any

Find an input-change sequence that reaches a target state.

For a target with more than one route — a Bool (Running, Running == False) or a word value (State == 5) — PILOT never reports ambiguous: it picks a deterministic default route, reaches the goal, and records where it went on Path.route. Exclude a reported route with avoid= and PILOT will look for another.

avoid X = do not take a path that depends on X. It excludes routes, operator actions, and observed scan states that satisfy the predicate — momentary commands are treated as actions, not just settled states (so avoid=C_Complete will not press C_Complete even though it settles back to rest). A path is excluded if it depends on the avoided condition at any of these three gates, and if every path is excluded the returned Path is unreachable with a reason that names the violated avoid condition(s).

Parameters:

Name Type Description Default
conditions Any

Target condition(s). Each is a Tag (Running) or a comparison (State == 3). Pass more than one for an AND goal — how(Running, State == 3) reaches a single state where every target holds, or reports why they can't coexist.

()
avoid Any

Condition(s) to keep the path (and the chosen route) clear of. One condition, or a tuple/list = union of exclusions (each condition avoided independently). Express a composite prohibition explicitly — avoid=And(A, B) avoids only the combined state, not A or B alone.

None
max_scans int

Scan budget for the search.

4000
unlink list[str] | None

Harness-feedback tag names to free for fault injection. Named tags have their link= coupling removed so the harness stops driving them and PILOT may steer them directly — the way to reach a fault (e.g. a dead sensor) that the intact physical link would otherwise hold out of reach.

None

Returns:

Name Type Description
A Any

class:~pyrung.core.analysis.graph.Path.

recovers

recovers(
    tag: Tag | str, *, assume: dict[str, Any] | None = None
) -> bool

True if tag has a reachable clear path from the current state.

Convenience predicate: cause(tag, to=resting).mode != 'unreachable'. For the underlying chain (witness or blockers), call cause() directly.

Tags marked external=True always return True — the recovery path exists outside the ladder by declaration. When assume is provided the external shortcut is skipped so the analysis runs with the given overrides.

Parameters:

Name Type Description Default
tag Tag | str

Tag object or tag name string.

required
assume dict[str, Any] | None

Tag-to-value overrides. Pins the given tags to specified values during projected analysis.

None

fork

fork(
    scan_id: int | None = None,
    *,
    history_budget: int | float | None = None,
    inherit_log: bool = True,
) -> PLC

Create an independent runner from retained historical state.

Parameters:

Name Type Description Default
scan_id int | None

Snapshot to fork from. Defaults to current committed tip state.

None
history_budget int | float | None

Byte ceiling for the child's recent-state cache. Defaults to the parent's budget. Pass math.inf to disable eviction entirely — every state the child produces stays cached, guaranteeing full-fidelity cause() chains without replay. Useful for ephemeral forks that will be discarded shortly after causal analysis.

None

fork_from

fork_from(scan_id: int) -> PLC

Create an independent runner from a retained historical snapshot.

replay_to

replay_to(target_scan_id: int) -> PLC

Reconstruct historical state, preferring compiled replay when supported.

replay_trace_at

replay_trace_at(
    target_scan_id: int,
) -> dict[int, RungTrace]

Reconstruct the rung-trace dict for a historical scan.

Runs the same replay walk as replay_to up to target_scan_id - 1 on the plain scan path, then drives _scan_steps_debug for target_scan_id so the replay fork's _current_rung_traces gets populated. Returns a copy of that dict; the replay fork is discarded.

The _replay_mode guards in _commit_scan (monitors, breakpoints) and _set_rtc_and_record cover the debug path too — both generators funnel through the same commit sink.

A one-slot cache (_cached_replay_trace) hits when the same target_scan_id is requested back-to-back. It is cleared on any tip advance (_run_single_scan) and on reset paths that reset the log (reboot, stop→run, via _clear_retained_debug_trace_caches).

Traces only exist for scans that actually executed — the fork anchor / initial scan was never stepped in debug mode — so target_scan_id must be strictly greater than _initial_scan_id.

stop

stop() -> None

Transition PLC to STOP mode.

reboot

reboot() -> SystemState

Power-cycle the runner and return the reset state.

Reboot is destructive: tags reset to defaults (except battery-preserved retentive values), state.scan_id and state.timestamp return to 0. Because post-reboot scan_ids would alias pre-reboot entries in every sparse channel (patches, forces, rtc_base_changes, dts), the scan log and checkpoints are reset to a fresh recording session rooted at the post-reboot scan 0. Pre-reboot history is not replay-addressable — users who need that should fork() before rebooting.

set_rtc

set_rtc(value: datetime) -> None

Set the current RTC value for the runner.

patch

patch(
    tags: Mapping[str, bool | int | float | str]
    | Mapping[Tag, bool | int | float | str]
    | Mapping[str | Tag, bool | int | float | str],
) -> None

Queue tag values for next scan (one-shot).

Values are applied at the start of the next step() call, then cleared. Use for momentary inputs like button presses.

Parameters:

Name Type Description Default
tags Mapping[str, bool | int | float | str] | Mapping[Tag, bool | int | float | str] | Mapping[str | Tag, bool | int | float | str]

Dict of tag names or Tag objects to values.

required

force

force(
    tag: str | Tag, value: bool | int | float | str
) -> None

Persistently override a tag value until explicitly removed.

The forced value is applied at the pre-logic force pass (phase 3) and re-applied at the post-logic force pass (phase 5) every scan. Logic may temporarily diverge the value mid-scan, but the post-logic pass restores it before outputs are written.

Forces persist across scans until unforce() or clear_forces() is called. Multiple forces may be active simultaneously.

If a tag is both patched and forced in the same scan, the force overwrites the patch during the pre-logic pass.

Parameters:

Name Type Description Default
tag str | Tag

Tag name or Tag object to override.

required
value bool | int | float | str

Value to hold. Must be compatible with the tag's type.

required

Raises:

Type Description
ValueError

If the tag is a read-only system point.

unforce

unforce(tag: str | Tag) -> None

Remove a single persistent force override.

After removal the tag resumes its logic-computed value starting from the next scan.

Parameters:

Name Type Description Default
tag str | Tag

Tag name or Tag object whose force to remove.

required

clear_forces

clear_forces() -> None

Remove all active persistent force overrides.

All forced tags resume their logic-computed values starting from the next scan.

forced

forced(
    overrides: Mapping[str, bool | int | float | str]
    | Mapping[Tag, bool | int | float | str]
    | Mapping[str | Tag, bool | int | float | str],
) -> Iterator[PLC]

Temporarily apply forces for the duration of the context.

On entry, saves the current force map and adds the given overrides. On exit (normally or on exception), the exact previous force map is restored — forces that existed before the context are reinstated, and forces added inside the context are removed.

Safe for nesting: inner forced() contexts layer on top of outer ones without disrupting them.

Parameters:

Name Type Description Default
overrides Mapping[str, bool | int | float | str] | Mapping[Tag, bool | int | float | str] | Mapping[str | Tag, bool | int | float | str]

Mapping of tag name / Tag object to forced value.

required
Example
with plc.forced({"AutoMode": True, "Fault": False}):
    plc.run(5)
# AutoMode and Fault forces released here

monitor

monitor(
    tag: str | Tag, callback: Callable[[Any, Any], None]
) -> _RunnerHandle

Call callback(current, previous) after commit when tag value changes.

when

when(
    *conditions: Condition
    | Tag
    | Callable[[SystemState], bool]
    | tuple[Condition | Tag, ...]
    | list[Condition | Tag],
) -> _BreakpointBuilder

Create a breakpoint builder evaluated after each committed scan.

Accepts Tag/Condition expressions (implicit AND) or a single callable predicate receiving SystemState.

step

step() -> SystemState

Execute one full scan cycle and return the committed state.

run

run(cycles: int) -> SystemState

Execute up to cycles scans, stopping early on pause breakpoints.

Parameters:

Name Type Description Default
cycles int

Number of scans to execute.

required

Returns:

Type Description
SystemState

The final SystemState after all cycles.

run_for

run_for(
    seconds: float, *, fold: bool = True
) -> SystemState

Run until simulation time advances by N seconds or a pause breakpoint fires.

Parameters:

Name Type Description Default
seconds float

Minimum simulation time to advance.

required
fold bool

When True (default), fold past timer/counter plateaus instead of stepping scan-by-scan.

True

Returns:

Type Description
SystemState

The final SystemState after reaching the target time.

run_until

run_until(
    *conditions: Condition
    | Tag
    | Callable[[SystemState], bool]
    | tuple[Condition | Tag, ...]
    | list[Condition | Tag],
    max_cycles: int = 10000,
    fold: bool = True,
) -> SystemState

Run until condition is true, pause breakpoint fires, or max_cycles reached.

Accepts Tag/Condition expressions (implicit AND) or a single callable predicate receiving SystemState.

Parameters:

Name Type Description Default
conditions Condition | Tag | Callable[[SystemState], bool] | tuple[Condition | Tag, ...] | list[Condition | Tag]

Condition expressions or a single callable predicate.

()
max_cycles int

Maximum scans before giving up (default 10000).

10000
fold bool

When True (default), fold past timer/counter plateaus instead of stepping scan-by-scan.

True

Returns:

Type Description
SystemState

The state that matched the condition, or final state if max reached.

pyrung.Physical dataclass

Declares physical feedback characteristics for a tag or field.

Bool dwell feedback (has timing)::

motor_fb = Physical("MotorFb", on_delay="2s", off_delay="500ms")

Profile-driven feedback (has a declarative spec)::

temp    = Physical("TempSensor", profile=Ramp(up=0.5, down=-0.05))
oven    = Physical("Oven",       profile=Approach(toward=180.0, rate=0.3))
encoder = Physical("Encoder",    profile=Pulse(on_dwell="8ms", off_dwell="8ms"))

The system field groups related feedback for reporting.

pyrung.Harness dataclass

Automatic feedback harness driven by Physical + link= declarations.

Walks all known tags to find link= couplings and lowers each to plant rungs scanned pre-logic (the input-read phase): a timing Physical becomes an on-delay/off-delay timer pair (bool dwell), and a declarative profile spec becomes calc/timer rungs — Ramp/Approach drive an analog register, Pulse drives a bool pulse train. No Python tick, no monitors.

Usage::

plc = PLC(logic, dt=0.010)
harness = Harness(plc)
harness.install()
plc.run_for(0.5)  # Fb synthesized automatically

pending_count property

pending_count: int

How many bool couplings are mid-transition this scan.

Under the dwell model there is no schedule heap; a bool coupling is "pending" while its feedback has not yet reached the value its currently sustained enable implies — i.e. an on/off-delay timer is still counting toward a Fb flip. PILOT's settle/coast (run_until pending_count == 0) advances scans until every bool feedback has caught up to its held command, exactly as draining the old heap did.

fork_onto

fork_onto(plc: PLC) -> Harness

Create a copy of this harness installed on plc.

The fork inherits the parent's committed state — including each bool timer's accumulator registers — so the dwell carries over with no re-seed; only the (stateless) timer instructions and the fractional remainder are copied. This is the seam the synthesis-overlay install grows into.

unlink(tags: list[str]) -> None

Remove couplings for the named feedback tags.

After unlinking, the Harness no longer synthesizes feedback for these tags — they become free inputs the caller can steer directly. Models a broken sensor or fault scenario.

couplings

couplings() -> Iterator[Coupling]

Iterate over all discovered couplings (bool and profile).

advance_profiles

advance_profiles() -> Iterator[Any]

Yield every harness-owned :class:AdvanceProfile.

This is the static reading of "En drives Fb toward a threshold" that PILOT's accumulator resolver consumes exactly like a timer's profile — so how(Fb >= threshold) learns "hold En, coast N scans" without running anything.

Bool couplings are real synthesized on/off-delay instructions. They do not belong to the user's :class:Program, so a program-only walk cannot see them; expose their own profiles here instead of making consumers understand synthesis-private accumulator names. An analog coupling is a per-scan profile tick with no instruction owner, so its adapter also lives here.

pyrung.Coupling dataclass

Public view of one enable→feedback coupling discovered by the harness.

pyrung.Ramp dataclass

Linear analog plant response — a constant-slope ramp.

Fb moves up units per second while the linked enable is active and down units per second otherwise (down is typically negative — an ambient decay or bleed-down; 0 means "hold on enable fall"). Rates are per second, so they are stable across scan periods.

Lowered by :func:pyrung.core.synthesis.analog_feedback_rungs to an arm latch plus two guarded calc plant rungs reading sys.dt — real, inspectable rungs that fold and trace natively (no Python tick, no io-gap).

The declaration is pure data, so it round-trips through a Click nickname comment as profile=ramp:up=<up>|down=<down> (see :meth:to_token).

to_token

to_token() -> str

The comment-field serialization: ramp:up=<up>|down=<down>.

pyrung.Approach dataclass

First-order (exponential) analog plant response.

Fb moves toward toward by fraction rate per second while the linked enable is active — Fb += rate*(toward - Fb)*dt — and holds on enable fall. toward is a constant setpoint (a number) or a tag name (a live setpoint register). Lowered to a single guarded calc plant rung reading sys.dt.

The slope depends on the current value, so how() measures the coast empirically (fork-and-run) rather than analytically. Round-trips through a Click comment as profile=approach:toward=<toward>|rate=<rate>.

to_token

to_token() -> str

The comment-field serialization: approach:toward=<toward>|rate=<rate>.

pyrung.Pulse dataclass

Bool pulse-train feedback — an astable oscillation while enabled.

Fb cycles high for on_dwell then low for off_dwell (compound duration strings, e.g. "0.5s") as long as the linked enable is active, and rests low when disabled. This is the declarative form of a discrete pulse sensor (a shaft encoder, a flow-meter pulse output) — lowered by :func:pyrung.core.synthesis.pulse_feedback_rungs to a self-resetting on-delay pair plus two toggle rungs (the classic ladder "flasher").

Round-trips through a Click comment as profile=pulse:on_dwell=<on_dwell>|off_dwell=<off_dwell>.

to_token

to_token() -> str

The comment-field serialization: pulse:on_dwell=<..>|off_dwell=<..>.

pyrung.system module-attribute

system = SystemNamespaces(
    sys=SysNamespace(
        always_on=Bool("sys.always_on"),
        first_scan=Bool("sys.first_scan"),
        scan_clock_toggle=Bool("sys.scan_clock_toggle"),
        clock_10ms=Bool("sys.clock_10ms"),
        clock_100ms=Bool("sys.clock_100ms"),
        clock_500ms=Bool("sys.clock_500ms"),
        clock_1s=Bool("sys.clock_1s"),
        clock_1m=Bool("sys.clock_1m"),
        clock_1h=Bool("sys.clock_1h"),
        mode_switch_run=Bool("sys.mode_switch_run"),
        mode_run=Bool("sys.mode_run"),
        cmd_mode_stop=Bool("sys.cmd_mode_stop"),
        cmd_watchdog_reset=Bool("sys.cmd_watchdog_reset"),
        fixed_scan_mode=Bool("sys.fixed_scan_mode"),
        battery_present=Bool("sys.battery_present"),
        scan_counter=Int(
            "sys.scan_counter", retentive=False
        ),
        scan_time_current_ms=Int(
            "sys.scan_time_current_ms", retentive=False
        ),
        scan_time_min_ms=Int(
            "sys.scan_time_min_ms", retentive=False
        ),
        scan_time_max_ms=Int(
            "sys.scan_time_max_ms", retentive=False
        ),
        scan_time_fixed_setup_ms=Int(
            "sys.scan_time_fixed_setup_ms", retentive=False
        ),
        interrupt_scan_time_ms=Int(
            "sys.interrupt_scan_time_ms", retentive=False
        ),
        dt=Real("sys.dt", retentive=False),
    ),
    rtc=RtcNamespace(
        year4=Int("rtc.year4", retentive=False),
        year2=Int("rtc.year2", retentive=False),
        month=Int("rtc.month", retentive=False),
        day=Int("rtc.day", retentive=False),
        weekday=Int("rtc.weekday", retentive=False),
        hour=Int("rtc.hour", retentive=False),
        minute=Int("rtc.minute", retentive=False),
        second=Int("rtc.second", retentive=False),
        new_year4=Int("rtc.new_year4", retentive=False),
        new_month=Int("rtc.new_month", retentive=False),
        new_day=Int("rtc.new_day", retentive=False),
        new_hour=Int("rtc.new_hour", retentive=False),
        new_minute=Int("rtc.new_minute", retentive=False),
        new_second=Int("rtc.new_second", retentive=False),
        apply_date=Bool("rtc.apply_date"),
        apply_date_error=Bool("rtc.apply_date_error"),
        apply_time=Bool("rtc.apply_time"),
        apply_time_error=Bool("rtc.apply_time_error"),
    ),
    fault=FaultNamespace(
        plc_error=Bool("fault.plc_error"),
        division_error=Bool("fault.division_error"),
        out_of_range=Bool("fault.out_of_range"),
        address_error=Bool("fault.address_error"),
        math_operation_error=Bool(
            "fault.math_operation_error"
        ),
        code=Int("fault.code", retentive=False),
    ),
    firmware=FirmwareNamespace(
        main_ver_low=Int(
            "firmware.main_ver_low", retentive=False
        ),
        main_ver_high=Int(
            "firmware.main_ver_high", retentive=False
        ),
        sub_ver_low=Int(
            "firmware.sub_ver_low", retentive=False
        ),
        sub_ver_high=Int(
            "firmware.sub_ver_high", retentive=False
        ),
    ),
    storage=StorageNamespace(
        sd=StorageSdNamespace(
            eject_cmd=Bool("storage.sd.eject_cmd"),
            delete_all_cmd=Bool(
                "storage.sd.delete_all_cmd"
            ),
            ready=Bool("storage.sd.ready"),
            write_status=Bool("storage.sd.write_status"),
            error=Bool("storage.sd.error"),
            error_code=Int(
                "storage.sd.error_code", retentive=False
            ),
        )
    ),
)