Runtime API
Tier: Stable Core
Runner lifecycle, system points, timebase helpers, and feedback harness support.
pyrung.PLC
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). |
dt |
float
|
Configured scan period in seconds. |
time_mode |
TimeMode
|
Current time mode (FIXED_STEP or REALTIME). |
bounds_violations
property
Constraint violations from the most recent scan, if any.
dataview
property
Chainable query over this program's tag dependency graph.
Convenience shorthand for plc.program.dataview() — builds
(and caches) the static program graph lazily on first access.
forces
property
Read-only view of active persistent overrides.
rewind
Move playhead backward in time by seconds and return snapshot.
rung_firings
Return rung firings for the given scan (default: playhead).
Returns PMap[int, PMap[str, Any]] mapping each rung index
that fired (had any write, even if all were filtered by PDG)
during the scan to the filtered {tag_name: value_written}
map. Synthesized on demand from per-rung range-encoded
timelines (:class:RungFiringTimelines); rungs with no
timeline covering scan_id contribute nothing.
Populated uniformly by both the non-debug (step() /
run()) and debug (DAP pyrungStepScan / continue) scan
paths via ScanContext.capturing_rung.
Values are attempted writes, including a later writer reasserting the already-pending value. Transition consumers compare committed state; causal execution consumers retain the exact writer occurrence.
diff
Return changed tag values between two retained historical scans.
cause
cause(
tag: Tag | str,
scan: int | None = None,
*,
to: Any = _SENTINEL,
assume: dict[str, Any] | None = None,
deep: bool = True,
since: int | None = None,
) -> CausalChain | None
Explain what caused a tag to transition.
Recorded (default, to omitted): walks recorded history
backward from the transition. Returns None if no transition
was found. With deep=True, each fired rung's steady enablers
are recursively explained through their establishing transitions
or an observed writer of their current value. Non-firing static
writers are not swept into the result. Terminals are classified in
:attr:~pyrung.core.analysis.causal.CausalChain.roots.
Projected (to=value): projects forward from the current
state, finding reachable paths that would drive the tag to value.
Returns a CausalChain with mode='projected' (reachable) or
mode='unreachable' (stranded, with blockers). Never
returns None in projected mode.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tag
|
Tag | str
|
Tag object or tag name string. |
required |
scan
|
int | None
|
Specific scan to examine (recorded mode only). |
None
|
to
|
Any
|
Target value for projected mode. When provided, the method returns the path that would drive tag to this value from the current state. |
_SENTINEL
|
assume
|
dict[str, Any] | None
|
Tag-to-value overrides for projected mode. Pins
the given tags to specified values during analysis.
Raises |
None
|
deep
|
bool
|
Recorded mode only. |
True
|
since
|
int | None
|
Recorded mode only. Inclusive earliest transition scan for an incident-local explanation. The state immediately before this scan is boundary context, not a cause to chase. Omit it to traverse the runner's complete inherited epoch lineage. |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
A |
CausalChain | None
|
class: |
CausalChain | None
|
(recorded mode only, when no transition was found). |
effect
effect(
tag: Tag | str,
scan: int | None = None,
*,
from_: Any = _SENTINEL,
to_value: Any = _SENTINEL,
assume: dict[str, Any] | None = None,
steady_state_k: int = 3,
max_scans: int = 1000,
) -> CausalChain | None
Trace the downstream effects of a tag transition.
Recorded (default, from_ omitted): walks recorded
history forward from an actual transition. Returns None if
no transition was found.
Projected (from_=value): what-if analysis — if the tag
transitioned from value right now, what downstream effects would
follow? Returns mode='projected' (possibly empty steps for
dead-end) or mode='unreachable' if the trigger can't fire.
Never returns None in projected mode.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tag
|
Tag | str
|
Tag object or tag name string. |
required |
scan
|
int | None
|
Specific scan of the transition (recorded mode only). |
None
|
from_
|
Any
|
Current value for projected what-if analysis. For
Bool tags the TO value is inferred as |
_SENTINEL
|
assume
|
dict[str, Any] | None
|
Tag-to-value overrides for projected mode. Pins
the given tags to specified values during analysis.
Raises |
None
|
steady_state_k
|
int
|
Stop after this many consecutive scans with no new effects (recorded mode only, default 3). |
3
|
max_scans
|
int
|
Hard cap on forward scans (recorded mode only, default 1000). |
1000
|
Returns:
| Name | Type | Description |
|---|---|---|
A |
CausalChain | None
|
class: |
CausalChain | None
|
(recorded mode only, when no transition was found). |
why
Explain how tags reached their current values from a snapshot.
No history required. Walks the program graph backward from each tag, using the current state as evidence. Terminates at external inputs.
Multiple tags produce one unified tree — a single explanation consistent with all observations.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tags
|
Tag | str
|
One or more Tag objects or tag name strings. |
()
|
Returns:
| Name | Type | Description |
|---|---|---|
A |
CausalChain
|
class: |
CausalChain
|
|
how
how(
*conditions: Any,
avoid: Any = None,
max_scans: int = 4000,
unlink: list[str] | None = None,
on_event: Any = None,
) -> Any
Find an input-change sequence that reaches a target state.
For a target with more than one route — a Bool (Running,
Running == False) or a word value (State == 5) — PILOT never
reports ambiguous: it picks a deterministic default route, reaches the
goal, and records where it went on Path.route. Exclude a reported
route with avoid= and PILOT will look for another.
avoid X = do not take a path that depends on X. It excludes routes,
operator actions, and observed scan states that satisfy the predicate —
momentary commands are treated as actions, not just settled states (so
avoid=C_Complete will not press C_Complete even though it
settles back to rest). A path is excluded if it depends on the avoided
condition at any of these three gates, and if every path is excluded
the returned Path is unreachable with a reason that names the violated
avoid condition(s).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conditions
|
Any
|
Target condition(s). Each is a Tag ( |
()
|
avoid
|
Any
|
Condition(s) to keep the path (and the chosen route) clear of.
One condition, or a tuple/list = union of exclusions (each
condition avoided independently). Express a composite prohibition
explicitly — |
None
|
max_scans
|
int
|
Scan budget for the search. |
4000
|
unlink
|
list[str] | None
|
Harness-feedback tag names to free for fault injection.
Named tags have their |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
A |
Any
|
class: |
recovers
True if tag has a reachable clear path from the current state.
Convenience predicate: cause(tag, to=resting).mode != 'unreachable'.
For the underlying chain (witness or blockers), call cause() directly.
Tags marked external=True always return True — the recovery path
exists outside the ladder by declaration. When assume is provided
the external shortcut is skipped so the analysis runs with the given
overrides.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tag
|
Tag | str
|
Tag object or tag name string. |
required |
assume
|
dict[str, Any] | None
|
Tag-to-value overrides. Pins the given tags to specified values during projected analysis. |
None
|
fork
fork(
scan_id: int | None = None,
*,
history_budget: int | float | None = None,
inherit_log: bool = True,
) -> PLC
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
|
history_budget
|
int | float | None
|
Byte ceiling for the child's recent-state cache.
Defaults to the parent's budget. Pass |
None
|
fork_from
Create an independent runner from a retained historical snapshot.
replay_to
Reconstruct historical state, preferring compiled replay when supported.
replay_trace_at
Reconstruct the rung-trace dict for a historical scan.
Runs the same replay walk as replay_to up to
target_scan_id - 1 on the plain scan path, then drives
_scan_steps_debug for target_scan_id so the replay
fork's _current_rung_traces gets populated. Returns a copy
of that dict; the replay fork is discarded.
The _replay_mode guards in _commit_scan (monitors,
breakpoints) and _set_rtc_and_record cover the debug path
too — both generators funnel through the same commit sink.
A one-slot cache (_cached_replay_trace) hits when the same
target_scan_id is requested back-to-back. It is cleared on
any tip advance (_run_single_scan) and on reset paths that
reset the log (reboot, stop→run, via
_clear_retained_debug_trace_caches).
Traces only exist for scans that actually executed — the fork
anchor / initial scan was never stepped in debug mode — so
target_scan_id must be strictly greater than
_initial_scan_id.
reboot
Power-cycle the runner and return the reset state.
Reboot is destructive: tags reset to defaults (except
battery-preserved retentive values), state.scan_id and
state.timestamp return to 0. Because post-reboot scan_ids
would alias pre-reboot entries in every sparse channel
(patches, forces, rtc_base_changes, dts), the scan log and
checkpoints are reset to a fresh recording session rooted at
the post-reboot scan 0. Pre-reboot history is not
replay-addressable — users who need that should fork()
before rebooting.
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 |
force
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 unforce() 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 |
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. |
unforce
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 |
required |
clear_forces
Remove all active persistent force overrides.
All forced tags resume their logic-computed values starting from the next scan.
forced
forced(
overrides: Mapping[str, bool | int | float | str]
| Mapping[Tag, bool | int | float | str]
| Mapping[str | Tag, bool | int | float | str],
) -> Iterator[PLC]
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 forced() 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 / |
required |
monitor
Call callback(current, previous) after commit when tag value changes.
when
when(
*conditions: Condition
| Tag
| Callable[[SystemState], bool]
| tuple[Condition | Tag, ...]
| list[Condition | Tag],
) -> _BreakpointBuilder
Create a breakpoint builder evaluated after each committed scan.
Accepts Tag/Condition expressions (implicit AND) or a single
callable predicate receiving SystemState.
run
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 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 |
fold
|
bool
|
When True (default), fold past timer/counter plateaus instead of stepping scan-by-scan. |
True
|
Returns:
| Type | Description |
|---|---|
SystemState
|
The final SystemState after reaching the target time. |
run_until
run_until(
*conditions: Condition
| Tag
| Callable[[SystemState], bool]
| tuple[Condition | Tag, ...]
| list[Condition | Tag],
max_cycles: int = 10000,
fold: bool = True,
) -> SystemState
Run until condition is true, pause breakpoint fires, or max_cycles reached.
Accepts Tag/Condition expressions (implicit AND) or a single
callable predicate receiving SystemState.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conditions
|
Condition | Tag | Callable[[SystemState], bool] | tuple[Condition | Tag, ...] | list[Condition | Tag]
|
Condition expressions or a single callable predicate. |
()
|
max_cycles
|
int
|
Maximum scans before giving up (default 10000). |
10000
|
fold
|
bool
|
When True (default), fold past timer/counter plateaus instead of stepping scan-by-scan. |
True
|
Returns:
| Type | Description |
|---|---|
SystemState
|
The state that matched the condition, or final state if max reached. |
pyrung.Physical
dataclass
Declares physical feedback characteristics for a tag or field.
Bool dwell feedback (has timing)::
motor_fb = Physical("MotorFb", on_delay="2s", off_delay="500ms")
Profile-driven feedback (has a declarative spec)::
temp = Physical("TempSensor", profile=Ramp(up=0.5, down=-0.05))
oven = Physical("Oven", profile=Approach(toward=180.0, rate=0.3))
encoder = Physical("Encoder", profile=Pulse(on_dwell="8ms", off_dwell="8ms"))
The system field groups related feedback for reporting.
pyrung.Harness
dataclass
Automatic feedback harness driven by Physical + link= declarations.
Walks all known tags to find link= couplings and lowers each to plant rungs
scanned pre-logic (the input-read phase): a timing Physical becomes an
on-delay/off-delay timer pair (bool dwell), and a declarative profile spec
becomes calc/timer rungs — Ramp/Approach drive an analog register,
Pulse drives a bool pulse train. No Python tick, no monitors.
Usage::
plc = PLC(logic, dt=0.010)
harness = Harness(plc)
harness.install()
plc.run_for(0.5) # Fb synthesized automatically
pending_count
property
How many bool couplings are mid-transition this scan.
Under the dwell model there is no schedule heap; a bool coupling is
"pending" while its feedback has not yet reached the value its
currently sustained enable implies — i.e. an on/off-delay timer is
still counting toward a Fb flip. PILOT's settle/coast (run_until
pending_count == 0) advances scans until every bool feedback has
caught up to its held command, exactly as draining the old heap did.
fork_onto
Create a copy of this harness installed on plc.
The fork inherits the parent's committed state — including each bool timer's accumulator registers — so the dwell carries over with no re-seed; only the (stateless) timer instructions and the fractional remainder are copied. This is the seam the synthesis-overlay install grows into.
unlink
Remove couplings for the named feedback tags.
After unlinking, the Harness no longer synthesizes feedback for these tags — they become free inputs the caller can steer directly. Models a broken sensor or fault scenario.
couplings
Iterate over all discovered couplings (bool and profile).
advance_profiles
Yield every harness-owned :class:AdvanceProfile.
This is the static reading of "En drives Fb toward a threshold" that
PILOT's accumulator resolver consumes exactly like a timer's profile —
so how(Fb >= threshold) learns "hold En, coast N scans" without
running anything.
Bool couplings are real synthesized on/off-delay instructions. They do
not belong to the user's :class:Program, so a program-only walk cannot
see them; expose their own profiles here instead of making consumers
understand synthesis-private accumulator names. An analog coupling is
a per-scan profile tick with no instruction owner, so its adapter also
lives here.
pyrung.Coupling
dataclass
Public view of one enable→feedback coupling discovered by the harness.
pyrung.Ramp
dataclass
Linear analog plant response — a constant-slope ramp.
Fb moves up units per second while the linked enable is active and
down units per second otherwise (down is typically negative — an
ambient decay or bleed-down; 0 means "hold on enable fall"). Rates are
per second, so they are stable across scan periods.
Lowered by :func:pyrung.core.synthesis.analog_feedback_rungs to an arm
latch plus two guarded calc plant rungs reading sys.dt — real,
inspectable rungs that fold and trace natively (no Python tick, no io-gap).
The declaration is pure data, so it round-trips through a Click nickname
comment as profile=ramp:up=<up>|down=<down> (see :meth:to_token).
pyrung.Approach
dataclass
First-order (exponential) analog plant response.
Fb moves toward toward by fraction rate per second while the
linked enable is active — Fb += rate*(toward - Fb)*dt — and holds on
enable fall. toward is a constant setpoint (a number) or a tag name (a
live setpoint register). Lowered to a single guarded calc plant rung
reading sys.dt.
The slope depends on the current value, so how() measures the coast
empirically (fork-and-run) rather than analytically. Round-trips through a
Click comment as profile=approach:toward=<toward>|rate=<rate>.
pyrung.Pulse
dataclass
Bool pulse-train feedback — an astable oscillation while enabled.
Fb cycles high for on_dwell then low for off_dwell (compound
duration strings, e.g. "0.5s") as long as the linked enable is active,
and rests low when disabled. This is the declarative form of a discrete
pulse sensor (a shaft encoder, a flow-meter pulse output) — lowered by
:func:pyrung.core.synthesis.pulse_feedback_rungs to a self-resetting
on-delay pair plus two toggle rungs (the classic ladder "flasher").
Round-trips through a Click comment as
profile=pulse:on_dwell=<on_dwell>|off_dwell=<off_dwell>.
pyrung.system
module-attribute
system = SystemNamespaces(
sys=SysNamespace(
always_on=Bool("sys.always_on"),
first_scan=Bool("sys.first_scan"),
scan_clock_toggle=Bool("sys.scan_clock_toggle"),
clock_10ms=Bool("sys.clock_10ms"),
clock_100ms=Bool("sys.clock_100ms"),
clock_500ms=Bool("sys.clock_500ms"),
clock_1s=Bool("sys.clock_1s"),
clock_1m=Bool("sys.clock_1m"),
clock_1h=Bool("sys.clock_1h"),
mode_switch_run=Bool("sys.mode_switch_run"),
mode_run=Bool("sys.mode_run"),
cmd_mode_stop=Bool("sys.cmd_mode_stop"),
cmd_watchdog_reset=Bool("sys.cmd_watchdog_reset"),
fixed_scan_mode=Bool("sys.fixed_scan_mode"),
battery_present=Bool("sys.battery_present"),
scan_counter=Int(
"sys.scan_counter", retentive=False
),
scan_time_current_ms=Int(
"sys.scan_time_current_ms", retentive=False
),
scan_time_min_ms=Int(
"sys.scan_time_min_ms", retentive=False
),
scan_time_max_ms=Int(
"sys.scan_time_max_ms", retentive=False
),
scan_time_fixed_setup_ms=Int(
"sys.scan_time_fixed_setup_ms", retentive=False
),
interrupt_scan_time_ms=Int(
"sys.interrupt_scan_time_ms", retentive=False
),
dt=Real("sys.dt", retentive=False),
),
rtc=RtcNamespace(
year4=Int("rtc.year4", retentive=False),
year2=Int("rtc.year2", retentive=False),
month=Int("rtc.month", retentive=False),
day=Int("rtc.day", retentive=False),
weekday=Int("rtc.weekday", retentive=False),
hour=Int("rtc.hour", retentive=False),
minute=Int("rtc.minute", retentive=False),
second=Int("rtc.second", retentive=False),
new_year4=Int("rtc.new_year4", retentive=False),
new_month=Int("rtc.new_month", retentive=False),
new_day=Int("rtc.new_day", retentive=False),
new_hour=Int("rtc.new_hour", retentive=False),
new_minute=Int("rtc.new_minute", retentive=False),
new_second=Int("rtc.new_second", retentive=False),
apply_date=Bool("rtc.apply_date"),
apply_date_error=Bool("rtc.apply_date_error"),
apply_time=Bool("rtc.apply_time"),
apply_time_error=Bool("rtc.apply_time_error"),
),
fault=FaultNamespace(
plc_error=Bool("fault.plc_error"),
division_error=Bool("fault.division_error"),
out_of_range=Bool("fault.out_of_range"),
address_error=Bool("fault.address_error"),
math_operation_error=Bool(
"fault.math_operation_error"
),
code=Int("fault.code", retentive=False),
),
firmware=FirmwareNamespace(
main_ver_low=Int(
"firmware.main_ver_low", retentive=False
),
main_ver_high=Int(
"firmware.main_ver_high", retentive=False
),
sub_ver_low=Int(
"firmware.sub_ver_low", retentive=False
),
sub_ver_high=Int(
"firmware.sub_ver_high", retentive=False
),
),
storage=StorageNamespace(
sd=StorageSdNamespace(
eject_cmd=Bool("storage.sd.eject_cmd"),
delete_all_cmd=Bool(
"storage.sd.delete_all_cmd"
),
ready=Bool("storage.sd.ready"),
write_status=Bool("storage.sd.write_status"),
error=Bool("storage.sd.error"),
error_code=Int(
"storage.sd.error_code", retentive=False
),
)
),
)