Program

pyrung.core.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 = PLCRunner(logic)

CountDownBuilder

Bases: _BuilderBase

Builder for count_down instruction with chaining API (Click-style).

Supports required .reset() chaining: count_down(done, acc, 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 (Click-style).

Supports optional .down() and required .reset() chaining: count_up(done, acc, preset=100).reset(reset_tag) count_up(done, acc, 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 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).

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

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.

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)

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.

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

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

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.

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 = PLCRunner(my_logic)

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`

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)

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

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.

Program

pyrung.core.program

CountDownBuilder

reset

CountUpBuilder

down

reset

OffDelayBuilder

OnDelayBuilder

reset

ShiftBuilder

clock

reset

Branch

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

Subroutine

SubroutineFunc

name property

ForbiddenControlFlowError

count_down

count_up

off_delay

on_delay

shift

all_of

Equivalent operator form:

any_of

Also works with | operator:

Grouped AND inside OR (explicit):

fall

rise

branch

forloop

subroutine

program

blockcopy

calc

call

Or with decorator:

copy

fill

latch

out

pack_bits

pack_text

pack_words

reset

return_early

run_enabled_function

run_function

search

unpack_to_bits

unpack_to_words

current `classmethod`

register_dialect `classmethod`

registered_dialects `classmethod`

comment `property` `writable`

name `property`