Skip to content

Runtime API

Tier: Stable Core

Runner lifecycle, system points, and timebase helpers.

pyrung.PLCRunner

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

current_state property 

current_state: SystemState

Current state snapshot.

history property 

history: History

Read-only history query surface.

playhead property 

playhead: int

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

simulation_time property 

simulation_time: float

Current simulation clock in seconds.

time_mode property 

time_mode: TimeMode

Current time mode.

system_runtime property 

system_runtime: SystemPointRuntime

System point runtime component.

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.

diff 

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

Return changed tag values between two retained historical scans.

inspect 

inspect(
    rung_id: int, scan_id: int | None = None
) -> RungTrace

Return retained rung-level debug trace for one scan.

If scan_id is omitted, the current playhead scan is inspected. Raises: KeyError: Missing scan id, or missing rung trace for retained scan.

inspect_event 

inspect_event() -> tuple[int, int, RungTraceEvent] | None

Return the latest debug-trace event for active/committed debug-path scans.

Returns:

Type Description
tuple[int, int, RungTraceEvent] | None

A tuple of (scan_id, rung_id, event). In-flight debug-scan events
tuple[int, int, RungTraceEvent] | None

are preferred when available. Otherwise, the latest retained committed
tuple[int, int, RungTraceEvent] | None

debug-scan event is returned.
Notes
  • This API is populated by scan_steps_debug() only.
  • Scans produced through step()/run()/run_for()/run_until() do not contribute trace events here.

fork 

fork(scan_id: int | None = None) -> PLCRunner

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

Create an independent runner from a retained historical snapshot.

stop 

stop() -> None

Transition PLC to STOP mode.

set_battery_present 

set_battery_present(value: bool) -> None

Configure simulated backup battery presence.

reboot 

reboot() -> SystemState

Power-cycle the runner and return the reset state.

set_rtc 

set_rtc(value: datetime) -> None

Set the current RTC value for the runner.

set_time_mode 

set_time_mode(mode: TimeMode, *, dt: float = 0.1) -> None

Set the time mode for simulation.

Parameters:

Name Type Description Default
mode TimeMode

TimeMode.FIXED_STEP or TimeMode.REALTIME.

 required
dt float

Time delta per scan (only used for FIXED_STEP mode).

 0.1

active 

active() -> Iterator[PLCRunner]

Bind this runner as active for live Tag.value access.

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

add_force 

add_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 remove_force() 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.

remove_force 

remove_force(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.

force 

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

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 force() 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 runner.force({"AutoMode": True, "Fault": False}):
    runner.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
    | tuple[Condition | Tag, ...]
    | list[Condition | Tag],
) -> _BreakpointBuilder

Create a condition breakpoint builder evaluated after each committed scan.

when_fn 

when_fn(
    predicate: Callable[[SystemState], bool],
) -> _BreakpointBuilder

Create a callable-predicate breakpoint builder evaluated after each scan.

scan_steps 

scan_steps() -> Generator[
    tuple[int, Rung, ScanContext], None, None
]

Execute one scan cycle and yield after each rung evaluation.

Scan phases: 1. Create ScanContext from current state 2. Apply pending patches to context 3. Apply persistent force overrides (pre-logic) 4. Calculate dt and inject into context 5. Evaluate all logic (writes batched in context), yielding after each rung 6. Re-apply force overrides (post-logic) 7. Batch _prev:* updates for edge detection 8. Commit all changes in single operation

The commit in phase 8 only happens when the generator is exhausted.

scan_steps_debug 

scan_steps_debug() -> Generator[ScanStep, None, None]

Execute one scan cycle and yield ScanStep objects at all boundaries.

Yields a ScanStep at each:

  • Top-level rung boundary (kind="rung")
  • Branch entry / exit (kind="branch")
  • Subroutine call and body steps (kind="subroutine")
  • Individual instruction boundaries (kind="instruction")

Each ScanStep carries source location metadata (source_file, source_line, end_line), rung enable state, and a trace of evaluated conditions and instructions.

This is the API used by the DAP adapter. Prefer scan_steps() for non-debug consumers — it has less overhead and a simpler yield type.

Note

Like scan_steps(), the scan is committed only when the generator is fully exhausted.

prepare_scan 

prepare_scan() -> tuple[ScanContext, float]

Debugger-facing scan preparation API.

commit_scan 

commit_scan(ctx: ScanContext, dt: float) -> None

Debugger-facing scan commit API.

iter_top_level_rungs 

iter_top_level_rungs() -> Iterable[Rung]

Debugger-facing top-level rung iterator.

evaluate_condition_value 

evaluate_condition_value(
    condition: Any, ctx: ScanContext
) -> tuple[bool, list[dict[str, Any]]]

Debugger-facing condition evaluation API.

condition_term_text 

condition_term_text(
    condition: Any, details: list[dict[str, Any]]
) -> str

Debugger-facing condition summary API.

condition_annotation 

condition_annotation(
    *, status: str, expression: str, summary: str
) -> str

Debugger-facing annotation API.

condition_expression 

condition_expression(condition: Any) -> str

Debugger-facing expression rendering API.

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
    | tuple[Condition | Tag, ...]
    | list[Condition | Tag],
    max_cycles: int = 10000,
) -> SystemState

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

Parameters:

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

Tag / Condition expressions evaluated with implicit AND.

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

run_until_fn 

run_until_fn(
    predicate: Callable[[SystemState], bool],
    *,
    max_cycles: int = 10000,
) -> SystemState

Run until callable predicate is true, paused, or max_cycles reached.

Parameters:

Name Type Description Default
predicate Callable[[SystemState], bool]

Callable receiving committed SystemState snapshots.

 required
max_cycles int

Maximum scans before giving up (default 10000).

 10000

Returns:

Type Description
SystemState

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

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

pyrung.TimeMode

Bases: Enum

Simulation time modes.

Each scan advances by a fixed dt, regardless of wall clock.

Use for unit tests and deterministic simulations.

Simulation clock tracks actual elapsed time.

Use for integration tests and hardware-in-loop.

pyrung.TimeUnit

Bases: Enum

Timer time units for Click PLC.

The accumulator stores integer values in the specified unit. Conversion from dt (seconds) uses appropriate scaling.

dt_to_units 

dt_to_units(dt_seconds: float) -> float

Convert dt in seconds to timer units (with fractional part).

pyrung.Tms module-attribute 

Tms = Tms

pyrung.Ts module-attribute 

Ts = Ts

pyrung.Tm module-attribute 

Tm = Tm

pyrung.Th module-attribute 

Th = Th

pyrung.Td module-attribute 

Td = Td