Runner

pyrung.core.runner

PLCRunner - Generator-driven PLC execution engine.

The runner orchestrates scan cycle execution with inversion of control. The consumer drives execution via step(), allowing input injection, inspection, and pause at any point.

Uses ScanContext to batch all tag/memory updates within a scan cycle, reducing object allocation from O(instructions) to O(1) per scan.

ScanStep `dataclass`

Debug scan step emitted at rung boundaries.

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.

Runner

pyrung.core.runner

ScanStep dataclass

PLCRunner

current_state property

history property

playhead property

simulation_time property

time_mode property

system_runtime property

forces property

seek

rewind

diff

inspect

inspect_event

fork

fork_from

stop

set_battery_present

reboot

set_rtc

set_time_mode

active

patch

add_force

remove_force

clear_forces

force

monitor

when

when_fn

scan_steps

scan_steps_debug

prepare_scan

commit_scan

iter_top_level_rungs

evaluate_condition_value

condition_term_text

condition_annotation

condition_expression

step

run

run_for

run_until

run_until_fn

ScanStep `dataclass`

current_state `property`

history `property`

playhead `property`

simulation_time `property`

time_mode `property`

system_runtime `property`

forces `property`