Core

pyrung.core

Immutable PLC Engine.

Redux-style architecture where logic is a pure function

Logic(Current_State) -> Next_State

Uses ScanContext for batched updates within a scan cycle, reducing object allocation from O(instructions) to O(1) per scan.

ScanContext

Batched write context for a single scan cycle.

Collects all tag and memory writes during a scan cycle, then commits them all at once to produce a new SystemState. Provides read-after-write visibility so subsequent instructions in the same scan see updated values.

Attributes:

Name	Type	Description
`_state`		The original SystemState (immutable, not modified).
`_tags_evolver`		Pyrsistent evolver for final tag commit.
`_memory_evolver`		Pyrsistent evolver for final memory commit.
`_tags_pending`	`dict[str, Any]`	Fast lookup dict for pending tag writes.
`_memory_pending`	`dict[str, Any]`	Fast lookup dict for pending memory writes.

scan_id `property`

scan_id: int

Current scan ID from the original state.

timestamp `property`

timestamp: float

Current timestamp from the original state.

original_state `property`

original_state: SystemState

Access to the original (unmodified) state.

Useful for operations that need to read original values, such as computing _prev:* for edge detection.

get_tag

get_tag(name: str, default: Any = None) -> Any

Get a tag value, checking pending writes first.

Provides read-after-write visibility within the same scan cycle.

Parameters:

Name	Type	Description	Default
`name`	`str`	The tag name to retrieve.	required
`default`	`Any`	Value to return if tag not found.	`None`

Returns:

Type	Description
`Any`	The tag value from pending writes, original state, or default.

get_memory

get_memory(key: str, default: Any = None) -> Any

Get a memory value, checking pending writes first.

Provides read-after-write visibility within the same scan cycle.

Parameters:

Name	Type	Description	Default
`key`	`str`	The memory key to retrieve.	required
`default`	`Any`	Value to return if key not found.	`None`

Returns:

Type	Description
`Any`	The memory value from pending writes, original state, or default.

set_tag

set_tag(name: str, value: Any) -> None

Set a tag value (batched, committed at end of scan).

Parameters:

Name	Type	Description	Default
`name`	`str`	The tag name to set.	required
`value`	`Any`	The value to set.	required

set_tags

set_tags(updates: dict[str, Any]) -> None

Set multiple tag values (batched, committed at end of scan).

Parameters:

Name	Type	Description	Default
`updates`	`dict[str, Any]`	Dict of tag names to values.	required

set_memory

set_memory(key: str, value: Any) -> None

Set a memory value (batched, committed at end of scan).

Parameters:

Name	Type	Description	Default
`key`	`str`	The memory key to set.	required
`value`	`Any`	The value to set.	required

set_memory_bulk

set_memory_bulk(updates: dict[str, Any]) -> None

Set multiple memory values (batched, committed at end of scan).

Parameters:

Name	Type	Description	Default
`updates`	`dict[str, Any]`	Dict of memory keys to values.	required

commit

commit(dt: float) -> SystemState

Commit all pending changes and advance to next scan.

Creates a new SystemState with all batched tag and memory updates, then advances scan_id and timestamp.

Parameters:

Name	Type	Description	Default
`dt`	`float`	Time delta in seconds to add to timestamp.	required

Returns:

Type	Description
`SystemState`	New SystemState with all changes applied.

RungTrace `dataclass`

Retained per-rung debug trace for one committed scan.

RungTraceEvent `dataclass`

One debug event captured for a rung during scan execution.

Expression

Bases: ABC

Base class for all mathematical expressions.

Expressions are lazy-evaluated at scan time against a ScanContext. They can be composed using arithmetic operators and compared to produce Conditions.

evaluate `abstractmethod`

evaluate(ctx: ScanContext) -> Numeric

Evaluate this expression against a ScanContext.

Parameters:

Name	Type	Description	Default
`ctx`	`ScanContext`	ScanContext for resolving tag values.	required

Returns:

Type	Description
`Numeric`	The numeric result of the expression.

Block `dataclass`

Factory for creating Tags from a typed memory region.

Block defines a named, 1-indexed memory region where every address shares the same TagType. Indexing a Block returns a cached LiveTag. The block holds no runtime values — all values live in SystemState.tags.

Address bounds are inclusive on both ends: Block("DS", INT, 1, 100) defines addresses 1–100 (100 tags). Indexing outside this range raises IndexError. Slice syntax (block[1:10]) is rejected — use .select(start, end) instead.

For sparse blocks (e.g. Click X/Y banks with non-contiguous valid addresses), pass valid_ranges to restrict which addresses within [start, end] are legal.

Parameters:

Name	Type	Description	Default
`name`	`str`	Block prefix used to generate tag names (e.g. `"DS"` → `"DS1"`, `"DS2"` …).	required
`type`	`TagType`	`TagType` shared by all tags in this block.	required
`start`	`int`	Inclusive lower bound address (must be ≥ 0).	required
`end`	`int`	Inclusive upper bound address (must be ≥ start).	required
`retentive`	`bool`	Whether tags in this block survive power cycles. Default `False`.	`False`
`valid_ranges`	`tuple[tuple[int, int], ...] \| None`	Optional tuple of `(lo, hi)` inclusive segments that constrain which addresses within `[start, end]` are accessible. Addresses outside all segments raise `IndexError`.	`None`
`address_formatter`	`Callable[[str, int], str] \| None`	Optional callable `(block_name, addr) → str` that overrides default tag name generation. Used by dialects for canonical display names like `"X001"`.	`None`

Example

DS = Block("DS", TagType.INT, 1, 100)
DS[1]          # → LiveTag("DS1", TagType.INT)
DS[101]        # → IndexError

# Range for block operations:
DS.select(1, 10)   # → BlockRange, tags DS1..DS10

# Indirect (pointer) addressing:
idx = Int("Idx")
DS[idx]        # → IndirectRef, resolved at scan time
DS[idx + 1]    # → IndirectExprRef

rename_slot

rename_slot(addr: int, name: str) -> None

Set the first-class logical name for one slot before materialization.

clear_slot_name

clear_slot_name(addr: int) -> None

Clear a first-class slot name override for one address.

configure_slot

configure_slot(
    addr: int,
    *,
    retentive: bool | None = None,
    default: object = UNSET,
) -> None

Set per-slot runtime policy before this slot is materialized.

configure_range

configure_range(
    start: int,
    end: int,
    *,
    retentive: bool | None = None,
    default: object = UNSET,
) -> None

Set per-slot policy for all valid addresses in the inclusive window.

clear_slot_config

clear_slot_config(addr: int) -> None

Clear per-slot policy overrides for one address.

clear_range_config

clear_range_config(start: int, end: int) -> None

Clear per-slot policy overrides for all valid addresses in a window.

slot_config

slot_config(addr: int) -> SlotConfig

Return the effective runtime slot policy without materializing a Tag.

select

select(start: int, end: int) -> BlockRange

select(
    start: Tag | Expression, end: int | Tag | Expression
) -> IndirectBlockRange

select(
    start: int, end: Tag | Expression
) -> IndirectBlockRange

select(
    start: int | Tag | Any, end: int | Tag | Any
) -> BlockRange | IndirectBlockRange

Select an inclusive range of addresses for block operations.

Both start and end are inclusive: DS.select(1, 10) yields 10 tags (1, 2, … 10). This mirrors the block constructor convention and avoids the off-by-one confusion of Python's half-open slices.

For sparse blocks (valid_ranges set), returns only the valid addresses within the window — gaps are silently skipped.

Parameters:

Name	Type	Description	Default
`start`	`int \| Tag \| Any`	Start address. `int` for a static range; `Tag` or `Expression` for a dynamically-resolved range.	required
`end`	`int \| Tag \| Any`	End address. `int` for a static range; `Tag` or `Expression` for a dynamically-resolved range.	required

Returns:

Type	Description
`BlockRange \| IndirectBlockRange`	`BlockRange` when both arguments are `int` (resolved at
`BlockRange \| IndirectBlockRange`	definition time). `IndirectBlockRange` when either argument is a
`BlockRange \| IndirectBlockRange`	`Tag` or `Expression` (resolved each scan at execution time).

Raises:

Type	Description
`ValueError`	If `start > end`.
`IndexError`	If either bound is outside the block's `[start, end]`.

Example

# Static range
DS.select(1, 100)              # BlockRange, DS1..DS100

# Sparse window (Click X bank)
x.select(1, 21)               # valid tags only: X001..X016, X021

# Dynamic range (resolved each scan)
DS.select(start_tag, end_tag)  # IndirectBlockRange

# Use with bulk instructions:
fill(0, DS.select(1, 10))
blockcopy(DS.select(1, 10), DD.select(1, 10))
search(">=", 100, DS.select(1, 100), result=Found, found=FoundFlag)

map_to

map_to(target: BlockRange) -> MappingEntry

Create a logical-to-hardware mapping entry.

BlockRange `dataclass`

Contiguous range of addresses for block operations.

Attributes:

Name	Type	Description
`block`	`Block`	Source Block.
`start`	`int`	Starting address (inclusive).
`end`	`int`	Ending address (inclusive).

addresses `property`

addresses: range | tuple[int, ...]

Return addresses in this block window, filtered by block rules.

reverse

reverse() -> BlockRange

Return this same window with address iteration reversed.

as_value

as_value() -> Any

Wrap this range for TXT->numeric character-value conversion.

as_ascii

as_ascii() -> Any

Wrap this range for TXT->numeric ASCII-code conversion.

IndirectBlockRange `dataclass`

Memory block with runtime-resolved bounds.

Wraps a Block with start/end that may be Tags or Expressions, resolved at scan time.

Attributes:

Name	Type	Description
`block`	`Block`	Source Block.
`start_expr`	`int \| Tag \| Any`	Start address (int, Tag, or Expression).
`end_expr`	`int \| Tag \| Any`	End address (int, Tag, or Expression).

resolve_ctx

resolve_ctx(ctx: ScanContext) -> BlockRange

Resolve expressions to concrete BlockRange using ScanContext.

reverse

reverse() -> IndirectBlockRange

Return this same dynamic window with address iteration reversed.

as_value

as_value() -> Any

Wrap this range for TXT->numeric character-value conversion.

as_ascii

as_ascii() -> Any

Wrap this range for TXT->numeric ASCII-code conversion.

IndirectExprRef `dataclass`

Tag with runtime-resolved address via expression.

IndirectExprRef wraps a Block and an Expression. The actual address is computed from the expression at scan time.

This enables pointer arithmetic like DS[idx + 1] where idx is a Tag.

Attributes:

Name	Type	Description
`block`	`Block`	Block to index into.
`expr`	`Any`	Expression whose value determines the address.

resolve_ctx

resolve_ctx(ctx: ScanContext) -> Tag

Resolve expression value to concrete Tag using ScanContext.

Parameters:

Name	Type	Description	Default
`ctx`	`ScanContext`	ScanContext to evaluate expression against.	required

Returns:

Type	Description
`Tag`	Concrete Tag at the computed address.

Raises:

Type	Description
`IndexError`	If resolved address is out of range.

IndirectRef `dataclass`

Tag with runtime-resolved address via pointer.

IndirectRef wraps a Block and pointer Tag. The actual address is resolved from the pointer's value at scan time.

Attributes:

Name	Type	Description
`block`	`Block`	Block to index into.
`pointer`	`Tag`	Tag whose value determines the address.

resolve

resolve(state: SystemState) -> Tag

Resolve pointer value to concrete Tag.

Parameters:

Name	Type	Description	Default
`state`	`SystemState`	Current system state to read pointer value from.	required

Returns:

Type	Description
`Tag`	Concrete Tag at the resolved address.

Raises:

Type	Description
`IndexError`	If resolved address is out of range.

resolve_ctx

resolve_ctx(ctx: ScanContext) -> Tag

Resolve pointer value to concrete Tag using ScanContext.

Parameters:

Name	Type	Description	Default
`ctx`	`ScanContext`	ScanContext to read pointer value from.	required

Returns:

Type	Description
`Tag`	Concrete Tag at the resolved address.

Raises:

Type	Description
`IndexError`	If resolved address is out of range.

InputBlock `dataclass`

Bases: Block

Factory for creating InputTag instances from a physical input memory region.

InputBlock is identical to Block except:

Indexing returns LiveInputTag (not LiveTag), so elements have .immediate.
Always non-retentive — physical inputs do not survive power cycles.

Use InputBlock when the tags represent real hardware inputs (sensors, switches, etc.). In simulation, values are supplied via runner.patch() or runner.add_force() during the Read Inputs scan phase.

Example

X = InputBlock("X", TagType.BOOL, 1, 16)
X[1]           # → LiveInputTag("X1", BOOL)
X[1].immediate # → ImmediateRef — bypass image table
X.select(1, 8) # → BlockRange for bulk operations

OutputBlock `dataclass`

Bases: Block

Factory for creating OutputTag instances from a physical output memory region.

OutputBlock is identical to Block except:

Indexing returns LiveOutputTag (not LiveTag), so elements have .immediate.
Always non-retentive — physical outputs do not survive power cycles.

Writes to OutputTag elements are immediately visible to subsequent rungs within the same scan (standard PLC behavior). The actual hardware write happens at the Write Outputs scan phase (phase 6).

Example

Y = OutputBlock("Y", TagType.BOOL, 1, 16)
Y[1]           # → LiveOutputTag("Y1", BOOL)
Y[1].immediate # → ImmediateRef — bypass image table
Y.select(1, 8) # → BlockRange for bulk operations

SlotConfig `dataclass`

Effective runtime policy for one block slot.

ForbiddenControlFlowError

Bases: RuntimeError

Raised when Python control flow is used inside strict DSL scope.

ForLoop

Context manager for a repeated instruction block within a rung.

Program

Container for PLC logic (rungs and subroutines).

Used as a context manager to capture rungs

with Program() as logic: with Rung(Button): out(Light)

Also works with PLCRunner

runner = PLCRunner(logic)

add_rung

add_rung(rung: Rung) -> None

Add a rung to the program or current subroutine.

start_subroutine

start_subroutine(name: str) -> None

Start defining a subroutine.

end_subroutine

end_subroutine() -> None

End subroutine definition.

call_subroutine

call_subroutine(
    name: str, state: SystemState
) -> SystemState

Execute a subroutine by name (legacy state-based API).

call_subroutine_ctx

call_subroutine_ctx(name: str, ctx: ScanContext) -> None

Execute a subroutine by name within a ScanContext.

current `classmethod`

current() -> Program | None

Get the current program context (if any).

register_dialect `classmethod`

register_dialect(
    name: str, validator: DialectValidator
) -> None

Register a portability validator callback for a dialect name.

registered_dialects `classmethod`

registered_dialects() -> tuple[str, ...]

Return registered dialect names in deterministic order.

validate

validate(
    dialect: str, *, mode: str = "warn", **kwargs: Any
) -> Any

Run dialect-specific portability validation for this Program.

evaluate

evaluate(ctx: ScanContext) -> None

Evaluate all main rungs in order (not subroutines) within a ScanContext.

Rung

Context manager for defining a rung.

Example

with Rung(Button): out(Light)

with Rung(Step == 0): out(Light1) copy(1, Step, oneshot=True)

comment `property` `writable`

comment: str | None

Rung comment (max 1400 chars).

SubroutineFunc

A decorated function that represents a subroutine.

Created by using @subroutine("name") as a decorator. When passed to call(), auto-registers with the current Program on first use.

Example

@subroutine("init") def init_sequence(): with Rung(): out(Light)

with Program() as logic: with Rung(Button): call(init_sequence)

name `property`

name: str

The subroutine name.

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.

SystemState

Bases: PRecord

Immutable snapshot of PLC state at a point in time.

Attributes:

Name	Type	Description
`scan_id`		Monotonically increasing scan counter.
`timestamp`		Simulation clock in seconds.
`tags`		Immutable mapping of tag names to values (bool, int, float, str).
`memory`		Immutable mapping for internal state (timers, counters, etc).

with_tags

with_tags(
    updates: dict[str, bool | int | float | str],
) -> SystemState

Return new state with updated tags. Original unchanged.

with_memory

with_memory(updates: dict[str, Any]) -> SystemState

Return new state with updated memory. Original unchanged.

next_scan

next_scan(dt: float) -> SystemState

Return new state for next scan cycle.

Parameters:

Name	Type	Description	Default
`dt`	`float`	Time delta in seconds to add to timestamp.	required

AutoDefault `dataclass`

Descriptor for per-instance numeric default sequences.

Field `dataclass`

Field metadata used by udt and named_array declarations.

InstanceView

1-based indexed view into one structure instance.

Bool `dataclass`

Bases: _TagTypeBase

Create a BOOL (1-bit boolean) tag.

Not retentive by default — resets to False on power cycle.

Example

Button = Bool("Button")
Light  = Bool("Light", retentive=True)

Char `dataclass`

Bases: _TagTypeBase

Create a CHAR (8-bit ASCII character) tag.

Retentive by default. Use for single-character text values. For multi-character strings, use a Block of CHAR tags. In the Click dialect, Txt is an alias for Char.

Example

ModeChar = Char("ModeChar")

Dint `dataclass`

Bases: _TagTypeBase

Create a DINT (32-bit signed integer, ±2 147 483 647) tag.

Retentive by default. Use for counters or values that exceed INT range.

Example

TotalCount = Dint("TotalCount")

ImmediateRef `dataclass`

Reference to the immediate (physical) value of an I/O tag.

Wraps an InputTag or OutputTag to access the physical I/O value directly, bypassing the scan-cycle image table.

tag `property`

tag: Tag

Backward-compatible alias for tag-wrapped immediate operands.

InputTag `dataclass`

Bases: Tag

Tag representing a physical input channel.

InputTag instances are produced exclusively by indexing an InputBlock. They add the .immediate property for bypassing the scan-cycle image table.

.immediate semantics by context:

Simulation (pure): validation-time annotation only; no runtime effect.
Click dialect: transcription hint for Click software export.
CircuitPython dialect: generates direct hardware-read code.
Hardware-in-the-loop: triggers a real hardware read mid-scan.

You cannot create an InputTag directly; use InputBlock[n] instead.

Example

X = InputBlock("X", TagType.BOOL, 1, 16)
sensor = X[3]          # LiveInputTag
sensor.immediate       # ImmediateRef — bypass image table

immediate `property`

immediate: ImmediateRef

Return an ImmediateRef that bypasses the input image table.

Int `dataclass`

Bases: _TagTypeBase

Create an INT (16-bit signed integer, −32768 to 32767) tag.

Retentive by default — survives power cycles.

Example

Step     = Int("Step")
preset = Int("preset", retentive=False)

OutputTag `dataclass`

Bases: Tag

Tag representing a physical output channel.

OutputTag instances are produced exclusively by indexing an OutputBlock. They add the .immediate property for bypassing the scan-cycle image table.

.immediate semantics by context:

Simulation (pure): validation-time annotation only; no runtime effect.
Click dialect: transcription hint for Click software export.
CircuitPython dialect: generates direct hardware-write code.
Hardware-in-the-loop: triggers a real hardware write mid-scan.

You cannot create an OutputTag directly; use OutputBlock[n] instead.

Example

Y = OutputBlock("Y", TagType.BOOL, 1, 16)
valve = Y[1]           # LiveOutputTag
valve.immediate        # ImmediateRef — bypass image table

immediate `property`

immediate: ImmediateRef

Return an ImmediateRef that bypasses the output image table.

Real `dataclass`

Bases: _TagTypeBase

Create a REAL (32-bit IEEE 754 float) tag.

Retentive by default. Use for analog presets and process values.

Example

Temperature = Real("Temperature")
FlowRate    = Real("FlowRate", retentive=False)

Tag `dataclass`

A reference to a value in SystemState.

Tags define what a value is (name, type, behavior) but hold no runtime state. Values live only in SystemState.tags.

Attributes:

Name	Type	Description
`name`	`str`	Unique identifier for this tag.
`type`	`TagType`	Data type (BOOL, INT, DINT, REAL, WORD, CHAR).
`default`	`Any`	Default value (None means use type default).
`retentive`	`bool`	Whether value survives power cycles.

value `property` `writable`

value: Any

Read or write this tag's value through the active runner scope.

Returns the current value as seen by the runner, including any pending patches or forces. Writes are staged as one-shot patches consumed at the next step().

Raises:

Type	Description
`RuntimeError`	If called outside `with runner.active(): ...`.

Example

runner = PLCRunner(logic)
with runner.active():
    print(StartButton.value)    # read current value
    StartButton.value = True    # queue for next scan

map_to

map_to(target: Tag) -> MappingEntry

Create a logical-to-hardware mapping entry.

as_value

as_value() -> Any

Wrap this tag for text->numeric character-value conversion.

as_ascii

as_ascii() -> Any

Wrap this tag for text->numeric ASCII-code conversion.

as_text

as_text(
    *,
    suppress_zero: bool = True,
    pad: int | None = None,
    exponential: bool = False,
    termination_code: int | str | None = None,
) -> Any

Wrap this tag for numeric->text conversion.

as_binary

as_binary() -> Any

Wrap this tag for numeric->text binary-copy conversion.

TagType

Bases: Enum

Data types for tags (IEC 61131-3 naming).

Word `dataclass`

Bases: _TagTypeBase

Create a WORD (16-bit unsigned integer, 0x0000–0xFFFF) tag.

Retentive by default. Use for bit-packed status registers or hex values. In the Click dialect, Hex is an alias for Word.

Example

StatusWord = Word("StatusWord")

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.

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

acos

acos(x: Expression | Numeric | Tag) -> MathFuncExpr

Arc cosine function (returns radians).

asin

asin(x: Expression | Numeric | Tag) -> MathFuncExpr

Arc sine function (returns radians).

atan

atan(x: Expression | Numeric | Tag) -> MathFuncExpr

Arc tangent function (returns radians).

cos

cos(x: Expression | Numeric | Tag) -> MathFuncExpr

Cosine function (radians).

degrees

degrees(x: Expression | Numeric | Tag) -> MathFuncExpr

Convert radians to degrees.

log

log(x: Expression | Numeric | Tag) -> MathFuncExpr

Natural logarithm.

log10

log10(x: Expression | Numeric | Tag) -> MathFuncExpr

Base-10 logarithm.

lro

lro(
    x: Expression | int | Tag, n: Expression | int | Tag
) -> ShiftFuncExpr

Rotate left function (16-bit): lro(value, count).

lsh

lsh(
    x: Expression | int | Tag, n: Expression | int | Tag
) -> ShiftFuncExpr

Left shift function: lsh(value, count).

radians

radians(x: Expression | Numeric | Tag) -> MathFuncExpr

Convert degrees to radians.

rro

rro(
    x: Expression | int | Tag, n: Expression | int | Tag
) -> ShiftFuncExpr

Rotate right function (16-bit): rro(value, count).

rsh

rsh(
    x: Expression | int | Tag, n: Expression | int | Tag
) -> ShiftFuncExpr

Right shift function: rsh(value, count).

sin

sin(x: Expression | Numeric | Tag) -> MathFuncExpr

Sine function (radians).

sqrt

sqrt(x: Expression | Numeric | Tag) -> MathFuncExpr

Square root function.

tan

tan(x: Expression | Numeric | Tag) -> MathFuncExpr

Tangent function (radians).

all_of

all_of(
    *conditions: Condition
    | Tag
    | ImmediateRef
    | tuple[Condition | Tag | ImmediateRef, ...]
    | list[Condition | Tag | ImmediateRef],
) -> AllCondition

AND condition - true when all sub-conditions are true.

This is equivalent to comma-separated rung conditions, but useful when building grouped condition trees with any_of() or &.

Example

with Rung(all_of(Ready, AutoMode)): out(StartPermissive)

Equivalent operator form:

with Rung((Ready & AutoMode) | RemoteStart): out(StartPermissive)

any_of

any_of(
    *conditions: Condition | Tag | ImmediateRef,
) -> AnyCondition

OR condition - true when any sub-condition is true.

Use this to combine multiple conditions with OR logic within a rung. Multiple conditions passed directly to Rung() are ANDed together.

Example

with Rung(Step == 1, any_of(Start, CmdStart)): out(Light) # True if Step==1 AND (Start OR CmdStart)

Also works with | operator:

with Rung(Step == 1, Start | CmdStart): out(Light)

Grouped AND inside OR (explicit):

with Rung(any_of(Start, all_of(AutoMode, Ready), RemoteStart)): out(Light)

Parameters:

Name	Type	Description	Default
`conditions`	`Condition \| Tag \| ImmediateRef`	Conditions to OR together.	`()`

Returns:

Type	Description
`AnyCondition`	AnyCondition that evaluates True if any sub-condition is True.

blockcopy

blockcopy(
    source: Any, dest: Any, oneshot: bool = False
) -> None

Block copy instruction.

Copies values from source BlockRange to dest BlockRange. Both ranges must have the same length.

Example

with Rung(CopyEnable): blockcopy(DS.select(1, 10), DD.select(1, 10))

Parameters:

Name	Type	Description	Default
`source`	`Any`	Source BlockRange or IndirectBlockRange from .select().	required
`dest`	`Any`	Dest BlockRange or IndirectBlockRange from .select().	required
`oneshot`	`bool`	If True, execute only once per rung activation.	`False`

branch

branch(*conditions: ConditionTerm) -> Branch

Create a parallel branch within a rung.

A branch executes when both the parent rung conditions AND the branch's own conditions are true.

Example

with Rung(Step == 0): out(Light1) with branch(AutoMode): # Only executes if Step==0 AND AutoMode out(Light2) copy(1, Step, oneshot=True)

Parameters:

Name	Type	Description	Default
`conditions`	`ConditionTerm`	Conditions that must be true (in addition to parent rung) for this branch's instructions to execute.	`()`

Returns:

Type	Description
`Branch`	Branch context manager.

calc

calc(
    expression: Any, dest: Tag, oneshot: bool = False
) -> Tag

Calc instruction.

Evaluates an expression and stores the result in dest, with truncation to the destination tag's bit width (modular wrapping).

Key differences from copy(): - Truncates result to destination tag's type width - Division by zero produces 0 (not infinity) - Infers decimal/hex behavior from referenced tag types

Example

with Rung(Enable): calc(DS1 * DS2 + DS3, Result) calc(MaskA & MaskB, MaskResult)

Parameters:

Name	Type	Description	Default
`expression`	`Any`	Expression, Tag, or literal to evaluate.	required
`dest`	`Tag`	Destination tag (type determines truncation width).	required
`oneshot`	`bool`	If True, execute only once per rung activation.	`False`

Returns:

Type	Description
`Tag`	The dest tag.

call

call(target: str | SubroutineFunc) -> None

Call a subroutine instruction.

Executes the named subroutine when the rung is true. Accepts either a string name or a @subroutine-decorated function.

Example

with Rung(Button): call("init_sequence")

with subroutine("init_sequence"): with Rung(): out(Light)

Or with decorator:

@subroutine("init") def init_sequence(): with Rung(): out(Light)

with Program() as logic: with Rung(Button): call(init_sequence)

copy

copy(
    source: Any,
    target: Tag | IndirectRef | IndirectExprRef,
    oneshot: bool = False,
) -> Tag | IndirectRef | IndirectExprRef

Copy instruction (CPY/MOV).

Copies source value to target.

Example

with Rung(Button): copy(5, StepNumber)

count_down

count_down(
    done_bit: Tag, accumulator: Tag, *, preset: Tag | int
) -> CountDownBuilder

Count Down instruction (CTD) - Click-style.

Creates a counter that decrements every scan while the rung condition is True. Use rise() on the condition for edge-triggered counting.

Example

with Rung(rise(Dispense)): count_down(done_bit, acc, preset=25).reset(Reload)

This is a terminal instruction. Requires .reset() chaining.

Parameters:

Name	Type	Description	Default
`done_bit`	`Tag`	Tag to set when accumulator <= -preset.	required
`accumulator`	`Tag`	Tag to decrement while rung condition is True.	required
`preset`	`Tag \| int`	Target value (Tag or int).	required

Returns:

Type	Description
`CountDownBuilder`	Builder for chaining .reset().

count_up

count_up(
    done_bit: Tag, accumulator: Tag, *, preset: Tag | int
) -> CountUpBuilder

Count Up instruction (CTU) - Click-style.

Creates a counter that increments every scan while the rung condition is True. Use rise() on the condition for edge-triggered counting.

Example

with Rung(rise(PartSensor)): count_up(done_bit, acc, preset=100).reset(ResetBtn)

This is a terminal instruction. Requires .reset() chaining.

Parameters:

Name	Type	Description	Default
`done_bit`	`Tag`	Tag to set when accumulator >= preset.	required
`accumulator`	`Tag`	Tag to increment while rung condition is True.	required
`preset`	`Tag \| int`	Target value (Tag or int).	required

Returns:

Type	Description
`CountUpBuilder`	Builder for chaining .down() and .reset().

fall

fall(tag: Tag | ImmediateRef) -> FallingEdgeCondition

Falling edge contact (FE).

True only on 1->0 transition. Requires PLCRunner to track previous values.

Example

with Rung(fall(Button)): reset(MotorRunning) # Resets when button is released

fill

fill(value: Any, dest: Any, oneshot: bool = False) -> None

Fill instruction.

Writes a constant value to every element in a BlockRange.

Example

with Rung(ClearEnable): fill(0, DS.select(1, 100))

Parameters:

Name	Type	Description	Default
`value`	`Any`	Value to write (literal, Tag, or Expression).	required
`dest`	`Any`	Dest BlockRange or IndirectBlockRange from .select().	required
`oneshot`	`bool`	If True, execute only once per rung activation.	`False`

forloop

forloop(count: Tag | int, oneshot: bool = False) -> ForLoop

Create a repeated instruction block context.

Example

with Rung(Enable): with forloop(10) as loop: copy(Source[loop.idx + 1], Dest[loop.idx + 1])

latch

latch(
    target: Tag | BlockRange | ImmediateRef,
) -> Tag | BlockRange | ImmediateRef

Latch/Set instruction (SET).

Sets target to True. Unlike OUT, does NOT reset when rung goes false. Use reset() to turn off.

Example

with Rung(StartButton): latch(MotorRunning) latch(C.select(1, 8))

off_delay

off_delay(
    done_bit: Tag,
    accumulator: Tag,
    *,
    preset: Tag | int,
    unit: TimeUnit = TimeUnit.Tms,
) -> OffDelayBuilder

Off-Delay Timer instruction (TOF) - Click-style.

Done bit is True while enabled. After disable, counts until preset, then done bit goes False. Auto-resets when re-enabled.

Example

with Rung(MotorCommand): off_delay(done_bit, acc, preset=10000)

Off-delay timers are composable in-rung (not terminal).

Parameters:

Name	Type	Description	Default
`done_bit`	`Tag`	Tag that stays True for preset time after rung goes false.	required
`accumulator`	`Tag`	Tag to increment while disabled.	required
`preset`	`Tag \| int`	Delay time in time units (Tag or int).	required
`unit`	`TimeUnit`	Time unit for accumulator (default: Tms).	`Tms`

Returns:

Type	Description
`OffDelayBuilder`	Builder for the off_delay instruction.

on_delay

on_delay(
    done_bit: Tag,
    accumulator: Tag,
    *,
    preset: Tag | int,
    unit: TimeUnit = TimeUnit.Tms,
) -> OnDelayBuilder

On-Delay Timer instruction (TON/RTON) - Click-style.

Accumulates time while rung is true.

Example

with Rung(MotorRunning): on_delay(done_bit, acc, preset=5000) # TON on_delay(done_bit, acc, preset=5000).reset(ResetBtn) # RTON

Without .reset(), this is TON and remains composable in-rung. With .reset(), this is RTON and becomes terminal in the current flow.

Parameters:

Name	Type	Description	Default
`done_bit`	`Tag`	Tag to set when accumulator >= preset.	required
`accumulator`	`Tag`	Tag to increment while enabled.	required
`preset`	`Tag \| int`	Target value in time units (Tag or int).	required
`unit`	`TimeUnit`	Time unit for accumulator (default: Tms).	`Tms`

Returns:

Type	Description
`OnDelayBuilder`	Builder for optional .reset() chaining.

out

out(
    target: Tag | BlockRange | ImmediateRef,
    oneshot: bool = False,
) -> Tag | BlockRange | ImmediateRef

Output coil instruction (OUT).

Sets target to True when rung is true. Resets to False when rung goes false.

Example

with Rung(Button): out(Light) out(Y.select(1, 4))

pack_bits

pack_bits(
    bit_block: Any, dest: Any, oneshot: bool = False
) -> None

Pack BOOL tags from a BlockRange into a register destination.

pack_text

pack_text(
    source_range: Any,
    dest: Any,
    *,
    allow_whitespace: bool = False,
    oneshot: bool = False,
) -> None

Pack Copy text mode: parse a TXT/CHAR range into a numeric destination.

pack_words

pack_words(
    word_block: Any, dest: Any, oneshot: bool = False
) -> None

Pack two 16-bit tags from a BlockRange into a 32-bit destination.

reset

reset(
    target: Tag | BlockRange | ImmediateRef,
) -> Tag | BlockRange | ImmediateRef

Reset/Unlatch instruction (RST).

Sets target to its default value (False for bits, 0 for ints).

Example

with Rung(StopButton): reset(MotorRunning) reset(C.select(1, 8))

return_early

return_early() -> None

Return from the current subroutine.

Example

with subroutine("my_sub"): with Rung(Abort): return_early()

rise

rise(tag: Tag | ImmediateRef) -> RisingEdgeCondition

Rising edge contact (RE).

True only on 0->1 transition. Requires PLCRunner to track previous values.

Example

with Rung(rise(Button)): latch(MotorRunning) # Latches on button press, not while held

run_enabled_function

run_enabled_function(
    fn: Callable[..., dict[str, Any]],
    ins: dict[
        str, Tag | IndirectRef | IndirectExprRef | Any
    ]
    | None = None,
    outs: dict[str, Tag | IndirectRef | IndirectExprRef]
    | None = None,
) -> None

Execute a synchronous function every scan with rung enabled state.

run_function

run_function(
    fn: Callable[..., dict[str, Any]],
    ins: dict[
        str, Tag | IndirectRef | IndirectExprRef | Any
    ]
    | None = None,
    outs: dict[str, Tag | IndirectRef | IndirectExprRef]
    | None = None,
    *,
    oneshot: bool = False,
) -> None

Execute a synchronous function when rung power is true.

search

search(
    condition: str,
    value: Any,
    search_range: BlockRange | IndirectBlockRange,
    result: Tag,
    found: Tag,
    continuous: bool = False,
    oneshot: bool = False,
) -> Tag

Search instruction.

Scans a selected range and writes the first matching address into result. Writes found True on hit; on miss writes result=-1 and found=False.

shift

shift(
    bit_range: BlockRange | IndirectBlockRange,
) -> ShiftBuilder

Shift register instruction builder.

Data input comes from current rung power. Use .clock(...) then .reset(...) to finalize and add the instruction.

Example

with Rung(DataBit): shift(C.select(2, 7)).clock(ClockBit).reset(ResetBit)

subroutine

subroutine(name: str, *, strict: bool = True) -> Subroutine

Define a named subroutine.

Subroutines are only executed when called via call(). They are NOT executed during normal program scan.

Example

with Program() as logic: with Rung(Button): call("my_sub")

with subroutine("my_sub"):
    with Rung():
        out(Light)

unpack_to_bits

unpack_to_bits(
    source: Any, bit_block: Any, oneshot: bool = False
) -> None

Unpack a register source into BOOL tags in a BlockRange.

unpack_to_words

unpack_to_words(
    source: Any, word_block: Any, oneshot: bool = False
) -> None

Unpack a 32-bit register source into two 16-bit tags in a BlockRange.

auto

auto(*, start: int = 1, step: int = 1) -> Any

Create a per-instance numeric default sequence descriptor.

named_array

named_array(
    base_type: object, *, count: int = 1, stride: int = 1
) -> Callable[[type[Any]], _NamedArrayRuntime]

Decorator that builds a single-type, instance-interleaved structured runtime.

udt

udt(
    *, count: int = 1, numbered: bool = False
) -> Callable[[type[Any]], _StructRuntime]

Decorator that builds a mixed-type structured runtime from annotations.

immediate

immediate(
    value: Tag | BlockRange | ImmediateRef,
) -> ImmediateRef

Wrap a tag or block range as an immediate operand.

Core

pyrung.core

ScanContext

scan_id property

timestamp property

original_state property

get_tag

get_memory

set_tag

set_tags

set_memory

set_memory_bulk

commit

RungTrace dataclass

RungTraceEvent dataclass

Expression

evaluate abstractmethod

Block dataclass

rename_slot

clear_slot_name

configure_slot

configure_range

clear_slot_config

clear_range_config

slot_config

select

map_to

BlockRange dataclass

addresses property

tags

reverse

as_value

as_ascii

IndirectBlockRange dataclass

resolve_ctx

reverse

as_value

as_ascii

IndirectExprRef dataclass

resolve_ctx

IndirectRef dataclass

resolve

resolve_ctx

InputBlock dataclass

OutputBlock dataclass

SlotConfig dataclass

ForbiddenControlFlowError

ForLoop

Program

add_rung

start_subroutine

end_subroutine

call_subroutine

call_subroutine_ctx

current classmethod

register_dialect classmethod

registered_dialects classmethod

validate

evaluate

Rung

comment property writable

SubroutineFunc

name property

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

scan_id `property`

timestamp `property`

original_state `property`

RungTrace `dataclass`

RungTraceEvent `dataclass`

evaluate `abstractmethod`

Block `dataclass`

BlockRange `dataclass`

addresses `property`

IndirectBlockRange `dataclass`

IndirectExprRef `dataclass`

IndirectRef `dataclass`

InputBlock `dataclass`

OutputBlock `dataclass`

SlotConfig `dataclass`

current `classmethod`

register_dialect `classmethod`

registered_dialects `classmethod`

comment `property` `writable`

name `property`

current_state `property`

history `property`

playhead `property`

simulation_time `property`

time_mode `property`

system_runtime `property`

forces `property`

AutoDefault `dataclass`

Field `dataclass`

Bool `dataclass`

Char `dataclass`

Dint `dataclass`

ImmediateRef `dataclass`

tag `property`

InputTag `dataclass`

immediate `property`

Int `dataclass`

OutputTag `dataclass`

immediate `property`

Real `dataclass`

Tag `dataclass`

value `property` `writable`

Word `dataclass`