Skip to content

Program Structure API

Tier: Stable Core

Program/rung builders and control-flow composition primitives.

pyrung.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 PLC

runner = PLC(logic)

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 | None = None,
    *,
    mode: str = "warn",
    select: set[str] | None = None,
    ignore: set[str] | None = None,
    dt: float = 0.01,
    **kwargs: Any,
) -> Any

Run validation on this Program.

Without arguments, runs all core validators and returns a ValidationReport. Use select / ignore to filter by rule code, and dt to configure the physical-realism validator.

With a dialect string, runs dialect-specific portability validation (e.g. logic.validate("click", mode="strict")).

dataview 

dataview() -> DataView

Return a chainable query over this program's tag dependency graph.

The graph is built lazily on first call and cached.

simplified 

simplified() -> dict[str, TerminalForm]

Compute the simplified Boolean form for every terminal tag.

pyrung.Rung

Context manager for defining a rung.

Example

with Rung(Button): out(Light)

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

continued 

continued() -> Rung

Mark this rung as a continuation of the previous rung's condition snapshot.

All conditions in this rung will evaluate against the same frozen state as the prior rung, rather than taking a fresh snapshot. This models multiple independent wires on the same visual rung in Click's ladder editor. The reused snapshot must come from the prior rung in the same execution scope (main or the current subroutine).

Returns:

Type Description
Rung

self, for chaining: with Rung(B).continued(): ...

pyrung.program

Program and Rung context managers for the immutable PLC engine.

Provides DSL syntax for building PLC programs:

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

runner = PLC(logic)

CountDownBuilder

Bases: _BuilderBase

Builder for count_down instruction with chaining API.

Supports required .reset() chaining: count_down(Counter[1], preset=25).reset(reset_tag)

reset 

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

Add reset condition (required).

When reset condition is true, loads preset into accumulator and clears done bit.

Parameters:

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

Condition(s) for resetting the counter.

 ()

Returns:

Type Description
Tag

The done bit tag.

CountUpBuilder

Bases: _BuilderBase

Builder for count_up instruction with chaining API.

Supports optional .down() and required .reset() chaining: count_up(Counter[1], preset=100).reset(reset_tag) count_up(Counter[1], preset=50).down(down_cond).reset(reset_tag)

down 

down(
    *conditions: Condition
    | Tag
    | tuple[Condition | Tag, ...]
    | list[Condition | Tag],
) -> CountUpBuilder

Add down trigger (optional).

Creates a bidirectional counter that increments on rung true and decrements on down condition true.

Parameters:

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

Condition(s) for decrementing the counter.

 ()

Returns:

Type Description
CountUpBuilder

Self for chaining.

reset 

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

Add reset condition (required).

When reset condition is true, clears both done bit and accumulator.

Parameters:

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

Condition(s) for resetting the counter.

 ()

Returns:

Type Description
Tag

The done bit tag.

OffDelayBuilder

Bases: _AutoFinalizeBuilderBase

Builder for off_delay instruction (TOF behavior, Click-style).

Auto-resets when re-enabled.

OnDelayBuilder

Bases: _AutoFinalizeBuilderBase

Builder for on_delay instruction with optional .reset() chaining (Click-style).

Without .reset(): TON behavior (auto-reset on rung false, non-terminal) With .reset(): RTON behavior (manual reset required, terminal)

reset 

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

Add reset condition (makes timer retentive - RTON).

When reset condition is true, clears both done bit and accumulator.

Parameters:

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

Condition(s) for resetting the timer.

 ()

Returns:

Type Description
Tag

The done bit tag.

ShiftBuilder

Bases: _BuilderBase

Builder for shift instruction with required .clock().reset() chaining.

clock 

clock(
    *conditions: Condition
    | Tag
    | tuple[Condition | Tag, ...]
    | list[Condition | Tag],
) -> ShiftBuilder

Set the shift clock trigger condition.

reset 

reset(
    *conditions: Condition
    | Tag
    | tuple[Condition | Tag, ...]
    | list[Condition | Tag],
) -> BlockRange | IndirectBlockRange

Finalize the shift instruction with required reset condition.

Branch

Context manager for 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)

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 PLC

runner = PLC(logic)

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 | None = None,
    *,
    mode: str = "warn",
    select: set[str] | None = None,
    ignore: set[str] | None = None,
    dt: float = 0.01,
    **kwargs: Any,
) -> Any

Run validation on this Program.

Without arguments, runs all core validators and returns a ValidationReport. Use select / ignore to filter by rule code, and dt to configure the physical-realism validator.

With a dialect string, runs dialect-specific portability validation (e.g. logic.validate("click", mode="strict")).

dataview 

dataview() -> DataView

Return a chainable query over this program's tag dependency graph.

The graph is built lazily on first call and cached.

simplified 

simplified() -> dict[str, TerminalForm]

Compute the simplified Boolean form for every terminal tag.

Rung

Context manager for defining a rung.

Example

with Rung(Button): out(Light)

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

continued 

continued() -> Rung

Mark this rung as a continuation of the previous rung's condition snapshot.

All conditions in this rung will evaluate against the same frozen state as the prior rung, rather than taking a fresh snapshot. This models multiple independent wires on the same visual rung in Click's ladder editor. The reused snapshot must come from the prior rung in the same execution scope (main or the current subroutine).

Returns:

Type Description
Rung

self, for chaining: with Rung(B).continued(): ...

Subroutine

Context manager for defining a subroutine.

Subroutines are named blocks of rungs that are only executed when called.

Example

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

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.

ForbiddenControlFlowError

Bases: RuntimeError

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

count_down 

count_down(
    counter: DoneAccUDT, preset: Tag | int
) -> CountDownBuilder

Count Down instruction (CTD).

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(Counter[1], preset=25).reset(Reload)

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

Parameters:

Name Type Description Default
counter DoneAccUDT

Counter instance (must have done and acc fields).

 required
preset Tag | int

Target value (Tag or int).

 required

Returns:

Type Description
CountDownBuilder

Builder for chaining .reset().

count_up 

count_up(
    counter: DoneAccUDT, preset: Tag | int
) -> CountUpBuilder

Count Up instruction (CTU).

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(Counter[1], preset=100).reset(ResetBtn)

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

Parameters:

Name Type Description Default
counter DoneAccUDT

Counter instance (must have done and acc fields).

 required
preset Tag | int

Target value (Tag or int).

 required

Returns:

Type Description
CountUpBuilder

Builder for chaining .down() and .reset().

off_delay 

off_delay(
    timer: DoneAccUDT,
    preset: Tag | int,
    unit: TimeUnitStr = "ms",
) -> OffDelayBuilder

Off-Delay Timer instruction (TOF).

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(Timer[2], preset=10000) off_delay(Timer[2], 10, "s") # 10 seconds

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

Parameters:

Name Type Description Default
timer DoneAccUDT

Timer instance (must have done and acc fields).

 required
preset Tag | int

Delay time in time units (Tag or int).

 required
unit TimeUnitStr

Time unit for accumulator (default: "ms"). Accepts "ms", "sec", "min", "hour", "day" and their variants (see :func:normalize_unit).

 'ms'

Returns:

Type Description
OffDelayBuilder

Builder for the off_delay instruction.

on_delay 

on_delay(
    timer: DoneAccUDT,
    preset: Tag | int,
    unit: TimeUnitStr = "ms",
) -> OnDelayBuilder

On-Delay Timer instruction (TON/RTON).

Accumulates time while rung is true.

Example

with Rung(MotorRunning): on_delay(Timer[1], preset=5000) # TON keyword on_delay(Timer[1], 5000) # TON positional on_delay(Timer[1], 5, "s") # TON 5 seconds on_delay(Timer[1], 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
timer DoneAccUDT

Timer instance (must have done and acc fields).

 required
preset Tag | int

Target value in time units (Tag or int).

 required
unit TimeUnitStr

Time unit for accumulator (default: "ms"). Accepts "ms", "sec", "min", "hour", "day" and their variants (see :func:normalize_unit).

 'ms'

Returns:

Type Description
OnDelayBuilder

Builder for optional .reset() chaining.

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)

And 

And(
    *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 Or().

Example

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

with Rung(Or(And(Ready, AutoMode), RemoteStart)): out(StartPermissive)

Or 

Or(
    *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, Or(Start, CmdStart)): out(Light) # True if Step==1 AND (Start OR CmdStart)

Grouped AND inside OR (explicit):

with Rung(Or(Start, And(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.

fall 

fall(tag: Tag | ImmediateRef) -> FallingEdgeCondition

Falling edge contact (FE).

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

Example

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

rise 

rise(tag: Tag | ImmediateRef) -> RisingEdgeCondition

Rising edge contact (RE).

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

Example

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

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.

comment 

comment(text: str) -> None

Attach a comment to the next rung.

The comment is consumed by the immediately following Rung() constructor. Must be called inside a Program context.

Example::

comment("UnitMode Change")
with Rung(C_UnitModeChgRequest):
    copy(1, C_UnitModeChgRequestBool, oneshot=True)

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

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)

program 

program(fn: Callable[[], None]) -> Program

program(
    fn: None = None, /, *, strict: bool = True
) -> Callable[[Callable[[], None]], Program]

program(
    fn: Callable[[], None] | None = None,
    /,
    *,
    strict: bool = True,
) -> Program | Callable[[Callable[[], None]], Program]

Decorator to create a Program from a function.

Example

@program def my_logic(): with Rung(Button): out(Light)

@program(strict=False) def permissive_logic(): with Rung(Button): out(Light)

runner = PLC(my_logic)

blockcopy 

blockcopy(
    source: Any,
    dest: Any,
    *,
    convert: CopyConverter | None = None,
    oneshot: bool = False,
) -> None

Block copy instruction.

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

Pass convert=to_value or convert=to_ascii for text→numeric block conversion. Only value and ascii modes are supported.

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
convert CopyConverter | None

Optional converter (to_value or to_ascii only).

 None
oneshot bool

If True, execute only once per rung activation.

 False

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,
    *,
    convert: CopyConverter | None = None,
    oneshot: bool = False,
) -> Tag | IndirectRef | IndirectExprRef

Copy instruction (CPY/MOV).

Copies source value to target. Pass convert= for text/numeric conversion (see :mod:pyrung.core.copy_converters).

Example

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

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

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

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

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(
    comparison: RangeComparison,
    *,
    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.

The first argument is a comparison expression built from a .select() range::

search(DS.select(1, 100) >= 100, result=FoundAddr, found=FoundFlag)
search(Txt.select(1, 50) == "A", result=FoundAddr, found=FoundFlag)

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.

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

pyrung.comment 

comment(text: str) -> None

Attach a comment to the next rung.

The comment is consumed by the immediately following Rung() constructor. Must be called inside a Program context.

Example::

comment("UnitMode Change")
with Rung(C_UnitModeChgRequest):
    copy(1, C_UnitModeChgRequestBool, oneshot=True)

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

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