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).
`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.

current_state `property`

current_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.

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.

.. todo::

A rung whose condition is True but whose writes are identical to
the already-pending values will not appear here.  This is an
acceptable approximation for causal-chain attribution; for
accurate cold-rung detection a ``_last_condition_result`` field
on ``Rung`` may be needed later.

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,
) -> 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.

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`

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,
    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).

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) -> 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`

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) -> 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

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,
) -> 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`

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 feedback (has timing)::

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

Analog feedback (has profile)::

temp = Physical("TempSensor", profile="first_order")

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, installs edge monitors on En tags, and schedules Fb patches using declared timing (bool) or profile functions (analog).

Usage::

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

couplings

couplings() -> Iterator[Coupling]

Iterate over all discovered couplings (bool and profile).

pyrung.Coupling `dataclass`

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

pyrung.profile

profile(name: str) -> Callable[..., Any]

Register an analog feedback profile function.

The decorated function is called once per scan tick for each active analog coupling::

@profile("generic_thermal")
def generic_thermal(cur, en, dt):
    if en:
        return cur + 0.5 * dt
    return cur

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
        ),
    ),
    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
            ),
        )
    ),
)

Runtime API

pyrung.PLC

program property

current_state property

tags property

bounds_violations property

history property

playhead property

dataview property

query property

simulation_time property

time_mode property

debug property

battery_present property writable

forces property

seek

rewind

rung_firings

diff

cause

effect

recovers

fork

fork_from

replay_to

replay_trace_at

stop

reboot

set_rtc

patch

force

unforce

clear_forces

forced

monitor

when

step

run

run_for

run_until

pyrung.Physical dataclass

pyrung.Harness dataclass

couplings

pyrung.Coupling dataclass

pyrung.profile

pyrung.system module-attribute

program `property`

current_state `property`

tags `property`

bounds_violations `property`

history `property`

playhead `property`

dataview `property`

query `property`

simulation_time `property`

time_mode `property`

debug `property`

battery_present `property` `writable`

forces `property`

pyrung.Physical `dataclass`

pyrung.Harness `dataclass`

pyrung.Coupling `dataclass`

pyrung.system `module-attribute`