"""WebSocket event parsing and dispatch.
Two transports — legacy ``/rest/subscribe`` (raw XML) and modern
``/api/events/subscribe`` (JSON-wrapped, adds PG3 ``spolisy`` channel)
— wrap the same ``<Event>`` payload, so :func:`parse_event_frame`
accepts either. System events use underscore-prefixed control ids
(``_5``, ``_28``, …) — see :class:`SystemEventControl`.
Decoupled from the WS reader (:mod:`pyisyox.runtime.ws`) so the
dispatcher can be tested with synthetic frames.
"""
from __future__ import annotations
import json
import logging
import re
from collections.abc import Callable
from dataclasses import dataclass
from datetime import UTC, datetime, tzinfo
from enum import IntEnum, StrEnum
from typing import TYPE_CHECKING
from xml.etree import ElementTree as ET
from pyisyox.client import NodePropertyValue
from pyisyox.constants import PROP_STATUS, SystemStatus
if TYPE_CHECKING:
from pyisyox.client import GroupRecord, NodeRecord, ProgramRecord, VariableRecord
_LOGGER = logging.getLogger(__name__)
def _enum_label(enum_cls: type[StrEnum], value: str) -> str:
"""Lower-case enum-member name for ``value``, or ``value`` verbatim
if it isn't a member. Shared by every ``Foo.label()`` classmethod."""
try:
return enum_cls(value).name.lower()
except ValueError:
return value
[docs]
class SystemEventControl(StrEnum):
"""IoX WebSocket "system" control codes (underscore-prefixed).
Property updates use the property id (``"ST"``, ``"GV1"``, ...) with a
populated ``node_address``. System events use one of these
underscore-prefixed codes with an empty ``node_address``.
Codes ``_0``-``_23`` are the full ISY-994 set from the *ISY994
Developer Cookbook* §8.5; ``_24``-``_28`` are IoX-6 additions
(system editors, the modern Z-Wave / ZigBee / Matter drivers,
system upgrade) not in that document — tracked from UDI's
internal ``UDEvents.h`` taxonomy. Newer IoX firmware may emit
further codes; those aren't enumerated, and :meth:`label` passes
them through verbatim so logs still identify them.
"""
#: Periodic heartbeat. ``<action>`` is the duration in seconds
#: until the next expected heartbeat (use it to detect a stalled
#: stream). No ``<eventInfo>``.
HEARTBEAT = "_0"
#: Trigger events — program status, variable change/init, schedule
#: change, key/info-string pushes, "get status" refresh signal.
#: ``<action>`` discriminates; see :class:`TriggerAction`.
TRIGGER = "_1"
#: Driver-specific events — payload depends on the underlying
#: protocol driver. Not modelled.
DRIVER_SPECIFIC = "_2"
#: Node / scene / folder lifecycle — add / remove / rename / enable
#: / revise / comm-error / etc. ``<action>`` carries the verb; see
#: :class:`NodeLifecycleAction` and :data:`NODE_LIFECYCLE_EVENT_INFO_TAGS`.
NODE_LIFECYCLE = "_3"
#: System configuration updated — time / NTP / notifications /
#: batch-mode / battery-write-mode. ``<action>`` 0-6; see
#: :class:`SystemConfigAction`.
SYSTEM_CONFIG = "_4"
#: Controller-side busy/idle/safe-mode status. ``<action>`` 0-3;
#: see :class:`pyisyox.constants.SystemStatus`.
SYSTEM_STATUS = "_5"
#: Internet-access status — disabled / enabled (``<eventInfo>`` =
#: external URL) / failed. See :class:`InternetAccessStatus`.
INTERNET_ACCESS = "_6"
#: Progress report during long-running operations (device
#: programming, restore, device-adder). ``<action>`` 1 / 2.1 / 2.2
#: / 2.3; see :class:`ProgressAction`. The ``_7A`` / ``_7M``
#: device-write sub-codes also ride through on this control — see
#: :class:`DeviceWriteAction`.
PROGRESS = "_7"
#: Security-system event — connected / disconnected / armed-* /
#: disarmed. See :class:`SecuritySystemAction`.
SECURITY_SYSTEM = "_8"
#: System alert event — "not implemented and should be ignored"
#: per the cookbook.
SYSTEM_ALERT = "_9"
#: OpenADR / Flex-Your-Power events — ISY994 Z-Series demand-response.
OPENADR = "_10"
#: Climate / weather events — required the ISY994 WeatherBug module;
#: not present on eisy.
CLIMATE = "_11"
#: AMI/SEP energy events — ISY994 only (see the Energy Management
#: Developer's Manual).
AMI_SEP = "_12"
#: External energy-monitoring (Brultech) — ISY994 only; on later
#: firmware these are folded into node events instead.
ENERGY_MONITORING = "_13"
#: UPB linker events — UPB-enabled units only.
UPB_LINKER = "_14"
#: UPB device-adder state — UPB-enabled units only.
UPB_DEVICE_ADDER = "_15"
#: UPB device-status events — UPB-enabled units only.
UPB_DEVICE_STATUS = "_16"
#: Gas-meter events — ISY994 only.
GAS_METER = "_17"
#: Legacy ZigBee events — ISY994-era driver. See :attr:`ZIGBEE_UYB`
#: (``_27``) for the IoX-6+ ZigBee driver used on eisy.
ZIGBEE = "_18"
#: ELK alarm-panel events — requires the ELK module (see the ELK
#: Integration Developer's Manual).
ELK = "_19"
#: Device-linker events — ``<action>`` 1 (status) / 2 (cleared).
#: See :class:`DeviceLinkerAction`.
DEVICE_LINKER = "_20"
#: Legacy Z-Wave integration events — ISY994-era driver. See
#: :attr:`ZMATTER_ZWAVE` (``_25``) for the IoX-6+ ZMatter Z-Wave
#: driver used on eisy.
ZWAVE = "_21"
#: Billing events — ISY994 ZS-series only.
BILLING = "_22"
#: Portal events — portal socket-connection / account-registration
#: status when a portal module is installed.
PORTAL = "_23"
#: System editor changed — fired when a "system editor" (e.g.
#: ``_sys_notify_short``) is updated. ``<node>`` carries the editor
#: name. ``<action>`` is :class:`SystemEditorAction`. IoX-6 addition.
SYSTEM_EDITOR = "_24"
#: ZMatter Z-Wave events — IoX-6+ Z-Wave driver on eisy hardware.
#: ``<action>`` is dotted (``"{category}.{type}"``); category
#: numbers are system-status (1), discovery (2), general-status (3),
#: general-error (4), S2 (5), OTA (6), backup/restore (7), device-
#: interview (8), button-detect (9), logger (10). Sub-action details
#: aren't modelled — the dotted string passes through verbatim.
#: Distinct from :attr:`ZWAVE` (``_21``, ISY994-era driver).
ZMATTER_ZWAVE = "_25"
#: System-upgrade lifecycle — ``<action>`` is :class:`SystemUpgradeAction`
#: (active / inactive / available / reboot-required). IoX-6 addition.
SYSTEM_UPGRADE = "_26"
#: ZigBee events — IoX-6+ ZigBee driver on eisy hardware. Same
#: dotted ``"{category}.{type}"`` action shape as :attr:`ZMATTER_ZWAVE`
#: (minus the logger sub-category). Distinct from :attr:`ZIGBEE`
#: (``_18``, ISY994-era driver).
ZIGBEE_UYB = "_27"
#: Matter network status — IoX-6+ Matter driver. ``<action>`` is
#: dotted; active sub-categories are 1 (system status), 2 (discovery),
#: 3 (RX/TX), 8 (device interview). Not in the ISY994 cookbook.
#:
#: **Note:** ``_28`` is also reserved in UDI's source for Profile
#: change events (actions 1-8 — profile/editor/nodedef/linkdef
#: updated/deleted) — but no firmware path fires those today
#: (placeholders since Dec 2024 per UDI). Don't subscribe expecting
#: them.
MATTER_STATUS = "_28"
[docs]
@classmethod
def label(cls, control: str) -> str:
"""Friendly name for a system control code, or the raw code if
unknown — so a log line reads ``node_lifecycle = ND`` instead
of ``_3 = ND``."""
return _enum_label(cls, control)
[docs]
class TriggerAction(StrEnum):
"""Action codes carried in :attr:`SystemEventControl.TRIGGER` (``_1``)
frames — *ISY994 Developer Cookbook* §8.5.3. ``<action>``
discriminates what the frame is; pyisyox only routes on
:attr:`PROGRAM_STATUS` / :attr:`VARIABLE_VALUE` / :attr:`VARIABLE_INIT`.
"""
#: Program status changed — handled by ``_apply_program_status``.
#: ``<eventInfo>`` carries the program ``<id>``, enabled/run-at-reboot
#: flags, last run/finish times, and a bitwise ``<s>`` status.
PROGRAM_STATUS = "0"
#: "Get status" — the controller is telling subscribers to re-poll
#: everything (e.g. after a config change). No payload.
GET_STATUS = "1"
#: A key changed. ``node`` carries the key.
KEY_CHANGED = "2"
#: An info string. ``node`` carries the key; ``<eventInfo>`` is the text.
INFO_STRING = "3"
#: IR learn mode toggled. No payload.
IR_LEARN_MODE = "4"
#: A schedule's status changed. ``node`` carries the key.
SCHEDULE = "5"
#: Variable value changed — handled by ``_apply_variable_change``.
#: ``<eventInfo>`` carries ``<var type id><val><ts>``.
VARIABLE_VALUE = "6"
#: Variable init (restore-on-startup) value changed — same handler /
#: payload shape as :attr:`VARIABLE_VALUE`, applied to ``init``.
VARIABLE_INIT = "7"
#: The current subscription key, sent once right after a new
#: subscription is established. ``<eventInfo>`` is the key.
KEY = "8"
#: Variable table structurally changed — fires when a variable is
#: added or removed, or when its precision is changed (a metadata
#: change that the per-value :attr:`VARIABLE_VALUE` /
#: :attr:`VARIABLE_INIT` frames don't cover). ``<eventInfo>``
#: carries ``<var><type>N</type><id>0</id></var>`` (id=0 is the
#: wildcard sentinel — "this whole type's table changed"). The
#: right response is to re-fetch ``/api/variables/{type}`` for the
#: affected type so the registry mirrors the controller's metadata
#: (precision in particular — the wire ``val`` from the per-value
#: frames is raw, and a stale precision will mis-render the value).
#: The dispatcher recognises it and fires
#: :meth:`EventDispatcher.add_variable_table_change_listener`
#: callbacks; the dispatcher itself does not re-fetch.
VARIABLE_TABLE_CHANGED = "9"
[docs]
@classmethod
def label(cls, value: str) -> str:
"""Friendly lower-case name for a trigger-action code, or the
raw value if it isn't one we know."""
return _enum_label(cls, value)
[docs]
class ProgressAction(StrEnum):
"""Action codes on :attr:`SystemEventControl.PROGRESS` (``_7``) frames
— *Cookbook* §8.5.9. ``<eventInfo>`` is free-text progress detail."""
#: Generic progress update.
UPDATE = "1"
#: Device-adder info (UPB only).
DEVICE_ADDER_INFO = "2.1"
#: Device-adder warning (UPB only).
DEVICE_ADDER_WARN = "2.2"
#: Device-adder error (UPB only).
DEVICE_ADDER_ERROR = "2.3"
[docs]
@classmethod
def label(cls, value: str) -> str:
"""Friendly lower-case name, or the raw value."""
return _enum_label(cls, value)
[docs]
class SystemConfigAction(StrEnum):
"""Action codes on :attr:`SystemEventControl.SYSTEM_CONFIG` (``_4``)
frames — *Cookbook* §8.5.6."""
TIME_CHANGED = "0"
TIME_CONFIG_CHANGED = "1"
NTP_SETTINGS_UPDATED = "2"
NOTIFICATIONS_SETTINGS_UPDATED = "3"
NTP_COMM_ERROR = "4"
#: Batch mode toggled — ``<eventInfo><status>`` is ``"1"``/``"0"``.
BATCH_MODE_UPDATED = "5"
#: Battery-powered-write mode toggled — ``<eventInfo><status>`` is
#: ``"1"``/``"0"``.
BATTERY_WRITE_MODE_UPDATED = "6"
[docs]
@classmethod
def label(cls, value: str) -> str:
"""Friendly lower-case name, or the raw value."""
return _enum_label(cls, value)
[docs]
class InternetAccessStatus(StrEnum):
"""Action codes on :attr:`SystemEventControl.INTERNET_ACCESS` (``_6``)
frames — *Cookbook* §8.5.8."""
DISABLED = "0"
#: Enabled — ``<eventInfo>`` is the external URL.
ENABLED = "1"
FAILED = "2"
[docs]
@classmethod
def label(cls, value: str) -> str:
"""Friendly lower-case name, or the raw value."""
return _enum_label(cls, value)
[docs]
class SecuritySystemAction(StrEnum):
"""Action codes on :attr:`SystemEventControl.SECURITY_SYSTEM` (``_8``)
frames — *Cookbook* §8.5.10. ``node`` and ``<eventInfo>`` are null."""
DISCONNECTED = "0"
CONNECTED = "1"
DISARMED = "DA"
ARMED_AWAY = "AW"
ARMED_STAY = "AS"
ARMED_STAY_INSTANT = "ASI"
ARMED_NIGHT = "AN"
ARMED_NIGHT_INSTANT = "ANI"
ARMED_VACATION = "AV"
[docs]
@classmethod
def label(cls, value: str) -> str:
"""Friendly lower-case name, or the raw value."""
return _enum_label(cls, value)
[docs]
class DeviceLinkerAction(StrEnum):
"""Action codes on :attr:`SystemEventControl.DEVICE_LINKER` (``_20``)
frames — *Cookbook* §8.5.22 (``udievnts.xsd``)."""
#: Linking status update — ``<eventInfo>`` carries device-linker info.
STATUS = "1"
#: The device-linking list was cleared. No payload.
CLEARED = "2"
[docs]
@classmethod
def label(cls, value: str) -> str:
"""Friendly lower-case name, or the raw value."""
return _enum_label(cls, value)
[docs]
class SystemUpgradeAction(StrEnum):
"""Action codes on :attr:`SystemEventControl.SYSTEM_UPGRADE` (``_26``)
frames — IoX-6 firmware-upgrade lifecycle."""
#: Upgrade in progress.
ACTIVE = "1"
#: Upgrade not active (post-completion or idle).
INACTIVE = "2"
#: A new upgrade is available to install.
AVAILABLE = "3"
#: Upgrade applied; reboot required to take effect.
REBOOT_REQUIRED = "4"
[docs]
@classmethod
def label(cls, value: str) -> str:
"""Friendly lower-case name, or the raw value."""
return _enum_label(cls, value)
[docs]
class SystemEditorAction(StrEnum):
"""Action codes on :attr:`SystemEventControl.SYSTEM_EDITOR` (``_24``)
frames. The ``<node>`` slot carries the editor name (e.g.
``_sys_notify_short``)."""
#: A system editor's contents changed.
EDITOR_CHANGED = "1"
[docs]
@classmethod
def label(cls, value: str) -> str:
"""Friendly lower-case name, or the raw value."""
return _enum_label(cls, value)
[docs]
class DeviceWriteAction(StrEnum):
"""Device-write sub-codes that ride through on ``_7``
(:attr:`SystemEventControl.PROGRESS`) frames — PyISY 3.x surfaced
these as ``NodeChangeAction.DEVICE_WRITING`` / ``DEVICE_MEMORY``.
Unlike the other action enums, these are *control-value* sub-codes
(they have the ``_`` prefix and arrive in the ``<control>`` slot),
not ``<action>`` values — the dispatcher doesn't route them; they
pass through as plain control events. ``<eventInfo>`` child tags per
code are in :data:`DEVICE_WRITE_PROGRESS_EVENT_INFO_TAGS`.
"""
#: Device-writing progress message — ``<eventInfo>`` carries
#: ``<message>``.
PROGRESS = "_7A"
#: Raw Insteon memory write — ``<eventInfo>`` carries ``<memory>`` /
#: ``<cmd1>`` / ``<cmd2>`` / ``<value>``. ``hacs-udi-iox``'s
#: backlight entities subscribe to this to catch memory-write echoes.
MEMORY = "_7M"
[docs]
@classmethod
def label(cls, value: str) -> str:
"""Friendly lower-case name, or the raw value."""
return _enum_label(cls, value)
[docs]
class NodeLifecycleAction(StrEnum):
"""Verbs the eisy emits via ``<control>_3</control>`` events —
*ISY994 Developer Cookbook* §8.5.5 ("Node Changed/Updated"). PyISY
3.x keeps the same mapping. ``<eventInfo>`` child tags per verb are
in :data:`NODE_LIFECYCLE_EVENT_INFO_TAGS`.
``EN`` carries an ``enabled`` boolean in ``<eventInfo>`` — there's
no separate "disabled" verb; the same code handles both transitions.
"""
# --- node verbs ---------------------------------------------------
#: Node added. ``<eventInfo>`` carries ``<nodeName>`` plus a
#: ``<nodeType>`` that is itself the full ``<node>`` element — see
#: :attr:`NodeLifecycleEvent.node_xml`.
NODE_ADDED = "ND"
#: Node removed (device deleted from the controller).
NODE_REMOVED = "NR"
#: Node renamed (display name changed).
NODE_RENAMED = "NN"
#: Node moved into a Scene.
NODE_MOVED = "MV"
#: Link changed (within a scene). **Not supported** by the
#: controller — kept for documentation; never observed.
LINK_CHANGED = "CL"
#: Node removed from a Scene.
NODE_REMOVED_FROM_GROUP = "RG"
#: Parent (primary node) changed.
PARENT_CHANGED = "PC"
#: Node enabled/disabled — direction is in ``eventInfo.enabled``.
NODE_ENABLED = "EN"
#: Power-info changed — ``<eventInfo>`` carries ``<deviceClass>`` /
#: ``<wattage>`` / ``<dcPeriod>``.
POWER_INFO_CHANGED = "PI"
#: Device ID changed. **Not implemented** by the controller — kept
#: for documentation.
DEVICE_ID_CHANGED = "DI"
#: Device property changed — UPB only.
DEVICE_PROPERTY_CHANGED = "DP"
#: Pending device operation queued, awaiting commit. On Insteon a
#: write (e.g. changing backlight level) surfaces ``WH`` first, then
#: :attr:`PROGRAMMING_DEVICE` (``WD``) while the value is written; a
#: property-update event arrives separately once it lands.
PENDING_DEVICE_OP = "WH"
#: The controller is carrying out a programming/write operation on
#: this node (follows :attr:`PENDING_DEVICE_OP`). Cookbook name:
#: "Programming Device". Not a completion signal — watch the
#: subsequent property-update event for the new value.
PROGRAMMING_DEVICE = "WD"
#: Node revised — drastically changed (UPB-style); the consumer
#: should discard cached info for the node and rebuild it.
#: ``<eventInfo>`` carries the full ``<node>`` structure.
NODE_REVISED = "RV"
#: Supported-type info changed — the node's nodedef assignment was
#: reassigned (e.g. a node server's ``changeNode``, or a device
#: driver detecting new capabilities). The primary signal that a
#: cached nodedef → entity mapping is stale. *Not* fired for
#: ``/rest/profiles`` definition updates or moves, or at startup
#: migration — those rewrite the profile DB without notifying.
NODE_TYPE_INFO_CHANGED = "NI"
#: All nodes for a single device have been added (bulk). Fired
#: after an include / re-pair so consumers can coalesce a single
#: refresh per device instead of per child node.
ALL_NODES_ADDED = "AA"
#: Scene link updated — a link's properties (on-level / ramp rate)
#: changed for an existing scene member.
LINK_UPDATED = "LU"
#: Discovering nodes (linking in progress). No node.
DISCOVERING_NODES = "SN"
#: Node discovery complete. No node.
NODE_DISCOVERY_COMPLETE = "SC"
#: Node communication error (device unreachable).
NODE_ERROR = "NE"
#: A previously-reported node communication error was cleared
#: (cookbook: "Clear Node Error / Comm. Errors Cleared") — the
#: companion to :attr:`NODE_ERROR`.
NODE_ERROR_CLEARED = "CE"
# --- folder verbs -------------------------------------------------
#: Folder added.
FOLDER_ADDED = "FD"
#: Folder removed.
FOLDER_REMOVED = "FR"
#: Folder renamed — ``<eventInfo>`` carries ``<newName>``.
FOLDER_RENAMED = "FN"
# --- scene/group verbs --------------------------------------------
#: Scene (group) added — ``<eventInfo>`` carries ``<groupName>`` /
#: ``<groupType>``.
GROUP_ADDED = "GD"
#: Scene (group) removed.
GROUP_REMOVED = "GR"
#: Scene (group) renamed — ``<eventInfo>`` carries ``<newName>``.
GROUP_RENAMED = "GN"
# --- networking verb ----------------------------------------------
#: A networking-module resource was renamed (``node`` = the new
#: name). Doesn't affect the node registry.
NET_RENAMED = "WR"
[docs]
@classmethod
def label(cls, value: str) -> str:
"""Friendly lower-case name for a lifecycle verb, or the raw
code if it isn't one we know."""
return _enum_label(cls, value)
#: ``<eventInfo>`` child element names carried by each lifecycle verb
#: (per the UDI notification table). An empty tuple means the frame
#: carries only the node address. Reference metadata for consumers that
#: want to parse the payload — pyisyox itself only parses the ``<node>``
#: element on ``NODE_ADDED`` (see ``NodeLifecycleEvent.node_xml``).
NODE_LIFECYCLE_EVENT_INFO_TAGS: dict[NodeLifecycleAction, tuple[str, ...]] = {
NodeLifecycleAction.NODE_ADDED: ("nodeName", "nodeType"), # <nodeType> is the full <node>
NodeLifecycleAction.NODE_REMOVED: (),
NodeLifecycleAction.NODE_RENAMED: ("newName",),
NodeLifecycleAction.NODE_MOVED: ("movedNode", "linkType"),
NodeLifecycleAction.LINK_CHANGED: (), # not supported
NodeLifecycleAction.NODE_REMOVED_FROM_GROUP: ("removedNode",),
NodeLifecycleAction.PARENT_CHANGED: ("node", "nodeType", "parent", "parentType"),
NodeLifecycleAction.NODE_ENABLED: ("enabled",),
NodeLifecycleAction.POWER_INFO_CHANGED: ("deviceClass", "wattage", "dcPeriod"),
NodeLifecycleAction.DEVICE_ID_CHANGED: (), # not implemented
NodeLifecycleAction.DEVICE_PROPERTY_CHANGED: (), # UPB only
NodeLifecycleAction.PENDING_DEVICE_OP: (),
NodeLifecycleAction.PROGRAMMING_DEVICE: (),
NodeLifecycleAction.NODE_REVISED: (), # plus the full <node> structure
NodeLifecycleAction.NODE_TYPE_INFO_CHANGED: (),
NodeLifecycleAction.ALL_NODES_ADDED: (),
NodeLifecycleAction.LINK_UPDATED: (),
NodeLifecycleAction.DISCOVERING_NODES: (),
NodeLifecycleAction.NODE_DISCOVERY_COMPLETE: (),
NodeLifecycleAction.NODE_ERROR: (),
NodeLifecycleAction.NODE_ERROR_CLEARED: (),
NodeLifecycleAction.FOLDER_ADDED: (),
NodeLifecycleAction.FOLDER_REMOVED: (),
NodeLifecycleAction.FOLDER_RENAMED: ("newName",),
NodeLifecycleAction.GROUP_ADDED: ("groupName", "groupType"),
NodeLifecycleAction.GROUP_REMOVED: (),
NodeLifecycleAction.GROUP_RENAMED: ("newName",),
NodeLifecycleAction.NET_RENAMED: (),
}
#: ``<eventInfo>`` child tags carried by each :class:`DeviceWriteAction`
#: control code. The dispatcher doesn't route these — reference metadata
#: for consumers that subscribe to ``_7A`` / ``_7M`` control events
#: directly.
DEVICE_WRITE_PROGRESS_EVENT_INFO_TAGS: dict[DeviceWriteAction, tuple[str, ...]] = {
DeviceWriteAction.PROGRESS: ("message",),
DeviceWriteAction.MEMORY: ("memory", "cmd1", "cmd2", "value"),
}
#: ``SystemEventControl`` member → the action-code enum that decodes
#: its ``<action>`` value (for the controls whose actions we model).
#: Drives :func:`describe_system_event`; consumers can use it directly
#: too (``_SYSTEM_ACTION_ENUMS.get(control)``).
_SYSTEM_ACTION_ENUMS: dict[SystemEventControl, type[StrEnum]] = {
SystemEventControl.TRIGGER: TriggerAction,
SystemEventControl.NODE_LIFECYCLE: NodeLifecycleAction,
SystemEventControl.SYSTEM_CONFIG: SystemConfigAction,
SystemEventControl.SYSTEM_STATUS: SystemStatus,
SystemEventControl.INTERNET_ACCESS: InternetAccessStatus,
SystemEventControl.PROGRESS: ProgressAction,
SystemEventControl.SECURITY_SYSTEM: SecuritySystemAction,
SystemEventControl.DEVICE_LINKER: DeviceLinkerAction,
SystemEventControl.SYSTEM_EDITOR: SystemEditorAction,
SystemEventControl.SYSTEM_UPGRADE: SystemUpgradeAction,
}
[docs]
def describe_system_event(control: str, action: str) -> str:
"""Render a ``<control>`` / ``<action>`` pair from a *system* event
frame as a friendly ``"<control_label> = <action_label>"`` string.
Resolves both halves to their enum names where one applies:
* ``"_5"`` / ``"0"`` → ``"system_status = not_busy"``
* ``"_1"`` / ``"0"`` → ``"trigger = program_status"``
* ``"_3"`` / ``"WH"`` → ``"node_lifecycle = pending_device_op"``
* ``"_4"`` / ``"5"`` → ``"system_config = batch_mode_updated"``
* ``"_8"`` / ``"AW"`` → ``"security_system = armed_away"``
* ``"_20"`` / ``"2"`` → ``"device_linker = cleared"``
* ``"_0"`` / ``"90"`` → ``"heartbeat = 90"`` (action = seconds to
the next heartbeat; not enumerated)
* ``"_28"`` / ``"1.3"`` → ``"matter_status = 1.3"`` (no enum)
* ``"_26"`` / ``"2"`` → ``"system_upgrade = inactive"``
* ``"_24"`` / ``"1"`` → ``"system_editor = editor_changed"``
* ``"_99"`` / ``"x"`` → ``"_99 = x"`` (control we don't recognise —
both halves pass through verbatim)
Intended for the debug logging consumers do over raw event frames
(so a line reads ``system_status = busy`` instead of
``system_status = 1``); not part of any dispatch path. Property-
update frames (non-underscore control) aren't system events — this
just echoes them back unchanged if you pass one.
"""
control_label = SystemEventControl.label(control)
try:
ctrl = SystemEventControl(control)
except ValueError:
return f"{control_label} = {action}"
action_enum = _SYSTEM_ACTION_ENUMS.get(ctrl)
action_label = _enum_label(action_enum, action) if action_enum is not None else action
return f"{control_label} = {action_label}"
def _scalar(text: str) -> str | bool:
"""Coerce an XML leaf's text for log rendering — ``"true"`` /
``"false"`` become real booleans (matches how the eisy means them);
everything else stays a string (no int-guessing — ``"007"`` ≠ ``7``)."""
low = text.lower()
if low == "true":
return True
if low == "false":
return False
return text
def _xml_to_obj(el: ET.Element) -> object:
"""Recursively turn an ``<eventInfo>`` child element into a
JSON-friendly value for human-readable logging.
Attributes and child elements share the dict's keyspace (the IoX
event schema doesn't collide names, and dropping an ``@`` prefix
reads cleaner); body text alongside children lands under ``#text``.
* leaf with text → the (scalar-coerced) text
* element with attributes and/or children → a dict
* empty self-closing element (``<on/>``, ``<nr/>``) → ``True`` — a
presence flag
"""
obj: dict[str, object] = {k: _scalar(v) for k, v in el.attrib.items()}
for child in el:
obj[child.tag] = _xml_to_obj(child)
text = (el.text or "").strip()
if text:
if obj:
obj["#text"] = _scalar(text)
else:
return _scalar(text)
if obj:
return obj
return True
def _compact_event_info(event_info: str) -> str | None:
"""Render an ``<eventInfo>`` payload as a compact, readable blob.
Well-formed XML fragments (variable / program / lifecycle / Matter /
Z-Wave payloads) become a JSON object — ``{"loglevel": "0",
"connected": true}`` instead of ``<loglevel>0</loglevel>
<connected>true</connected>``. Non-XML payloads (``_7`` controller
logs carry CDATA, the subscription key is a bare string) fall back
to whitespace-collapsed text. ``None`` when there's nothing to show.
"""
if not event_info:
return None
try:
root = ET.fromstring(f"<eventInfo>{event_info}</eventInfo>") # noqa: S314 — eisy LAN traffic
except ET.ParseError:
collapsed = " ".join(event_info.split())
return collapsed or None
obj = {child.tag: _xml_to_obj(child) for child in root}
if obj:
return json.dumps(obj, default=str)
text = (root.text or "").strip()
return " ".join(text.split()) or None
def _log_system_event(event: Event) -> None:
"""``DEBUG`` line for a *system* event the dispatcher doesn't route
to its own handler — heartbeats, ``system_status`` / config toggles,
the ``_1`` trigger sub-events we don't act on (key / info-string /
get-status / schedule), and the protocol-driver / billing / Matter /
Z-Wave / portal frames.
``describe_system_event`` supplies the friendly ``control = action``
label; :func:`_compact_event_info` adds the payload as a JSON blob
when there is one. Routed frames (lifecycle, program-status,
variable-change) get their own purpose-built line from the handler
that decodes them — see :meth:`EventDispatcher._emit_lifecycle` etc.
Callers gate this on ``_LOGGER.isEnabledFor(DEBUG)``.
"""
if event.control == SystemEventControl.HEARTBEAT:
_LOGGER.debug("ISY heartbeat (next within %ss)", event.action or "?")
return
parts = [describe_system_event(event.control, event.action)]
if event.node_address:
parts.append(f"node={event.node_address}")
extra = _compact_event_info(event.event_info)
if extra:
parts.append(extra)
_LOGGER.debug("System event: %s", " ".join(parts))
[docs]
@dataclass(slots=True, frozen=True)
class NodeLifecycleEvent:
"""A high-level summary of a ``<control>_3</control>`` lifecycle frame.
Emitted alongside the raw :class:`Event` whenever the dispatcher
sees one of the actions in :class:`NodeLifecycleAction`. Consumers
subscribe via
:meth:`pyisyox.controller.Controller.add_node_lifecycle_listener`
to drive their own reload UX (HA Core's Repair issue, etc.).
Attributes:
action: The lifecycle verb (typed enum). Unknown verbs come
through as a plain string via :attr:`raw_action`.
node_address: Wire address of the affected node. Empty
string only for system-wide signals (none observed yet).
raw_action: The string action value verbatim, in case a new
verb appears that isn't yet in :class:`NodeLifecycleAction`.
seqnum: Sequence number of the underlying :class:`Event`.
node_xml: For ``ND`` actions, the inner ``<node>`` element
text from ``<eventInfo>``. ``None`` for verbs that don't
include the full element. Consumers wanting the parsed
shape can pass this to :func:`parse_lifecycle_node_xml`.
enabled: For ``EN`` (``NODE_ENABLED``) actions, the new
enabled/disabled state from ``<eventInfo><enabled>`` — the
same value already written back to ``Node.enabled``. ``None``
for every other verb (and for ``EN`` frames that omit the
flag).
"""
action: NodeLifecycleAction | str
node_address: str
raw_action: str
seqnum: int
node_xml: str | None = None
enabled: bool | None = None
@property
def requires_reload(self) -> bool:
"""True for verbs that invalidate the cached node/group/folder registry.
Reload-worthy: ``ND`` / ``NR`` / ``NN`` (node added/removed/renamed
— the registry's set or display names are stale), ``EN``
(enabled/disabled — the entity's property shape may change),
``RV`` (revised — discard and rebuild this node), ``NI``
(supported-type info changed — the node's nodedef assignment was
reassigned, so the cached nodedef→entity mapping is stale; per
UDI's notification taxonomy this is *the* primary signal for
profile-related node changes), ``AA`` (all-nodes-added bulk
signal after a device include), ``RG`` (removed from scene —
membership changed), ``SC`` (node-discovery complete — new
nodes may have appeared), and the folder/scene tree verbs
``FD`` / ``FR`` / ``FN`` / ``GD`` / ``GR`` / ``GN`` (the
``groups`` / ``folders`` registries are stale).
Softer signals — informational, don't trigger reload UX:
``MV`` (added to scene), ``CL`` (link changed — not supported),
``LU`` (scene link's on-level/ramp updated — property change,
not shape change), ``PC`` (parent changed), ``PI`` (power
info), ``DI`` (device id — not implemented), ``DP`` (UPB
property), ``WH`` (pending op), ``WD`` (programming device — a
property-update event follows), ``SN`` (discovering nodes —
wait for ``SC``), ``CE`` / ``NE`` (comm error/cleared — no
shape change), ``WR`` (a networking resource was renamed —
doesn't touch nodes).
"""
return self.action in {
NodeLifecycleAction.NODE_ADDED,
NodeLifecycleAction.NODE_REMOVED,
NodeLifecycleAction.NODE_RENAMED,
NodeLifecycleAction.NODE_REMOVED_FROM_GROUP,
NodeLifecycleAction.NODE_ENABLED,
NodeLifecycleAction.NODE_REVISED,
NodeLifecycleAction.NODE_TYPE_INFO_CHANGED,
NodeLifecycleAction.ALL_NODES_ADDED,
NodeLifecycleAction.NODE_DISCOVERY_COMPLETE,
NodeLifecycleAction.FOLDER_ADDED,
NodeLifecycleAction.FOLDER_REMOVED,
NodeLifecycleAction.FOLDER_RENAMED,
NodeLifecycleAction.GROUP_ADDED,
NodeLifecycleAction.GROUP_REMOVED,
NodeLifecycleAction.GROUP_RENAMED,
}
[docs]
@dataclass(slots=True, frozen=True)
class Event:
"""One parsed event frame.
Attributes:
seqnum: Event sequence number from the eisy. Monotonic per
connection; resets on reconnect.
timestamp: ISO 8601 timestamp string from the frame
(preserved verbatim — consumer parses if needed).
control: Property id (``"ST"``, ``"GV1"``, ...) or system code
(``"_5"``, ``"_28"``, ...).
action: Raw value as reported (string form preserves the
controller's precision representation).
node_address: Wire address of the affected node, or empty
string for system events.
formatted_action: Human-readable display value (e.g.
``"0.6839 US gallons"``). Empty when the controller didn't
supply one (system events typically don't).
formatted_name: Display name of the property (e.g.
``"Current"``). Empty when not provided.
uom: Unit-of-measure id from ``<action uom="...">``.
precision: Decimal precision from ``<action prec="...">``, or
``None`` if absent. (Wire keys it as ``"prec"``; Python
attribute spells it out.)
event_info: Inner ``<eventInfo>`` XML preserved verbatim.
Empty string when the frame had no ``<eventInfo>`` element
or when its content was empty. Consumers that need the
structured payload (e.g. variable change frames carrying
``<var type="..." id="...">``, or controller logs in CDATA)
parse this themselves — the IoX wire schema differs across
system control codes and pyisyox stays neutral.
"""
seqnum: int
timestamp: str
control: str
action: str
node_address: str
formatted_action: str = ""
formatted_name: str = ""
uom: str = ""
precision: int | None = None
event_info: str = ""
@property
def is_system(self) -> bool:
"""True for system control codes (``_5``, ``_28``, ...)."""
return self.control.startswith("_")
@property
def is_node_property(self) -> bool:
"""True when this event should overlay onto a node's property dict."""
return not self.is_system and bool(self.node_address) and bool(self.control)
[docs]
def parse_event_frame(raw: str) -> Event | None:
"""Decode a single WebSocket frame to an :class:`Event`.
Accepts either:
* Raw XML — ``<?xml...?><Event...>...</Event>`` (legacy
``/rest/subscribe``).
* JSON envelope — ``{"type": "event", "data": "<xml>"}`` (modern
``/api/events/subscribe``). Other ``type`` values (e.g.
``"spolisy"`` PG3 service status) return ``None`` — they're not
property updates and the dispatcher ignores them.
Returns ``None`` for keep-alive nulls, malformed XML, or non-event
JSON envelopes. Does **not** raise on parse failures so a single
bad frame can't crash the read loop.
"""
if not raw:
return None
payload = _maybe_unwrap_json_envelope(raw)
if payload is None:
return None
try:
root = ET.fromstring(payload) # noqa: S314 — eisy LAN traffic
except ET.ParseError as exc:
# eisy occasionally emits frames with unescaped ``&`` in text
# content — typically inside an ``_7`` REST-log ``<eventInfo>``
# echo of a ``submitCmd`` query string like
# ``NUM.107=24&VAL.111=1`` — which is not well-formed XML. Try
# again after escaping stray ampersands; suppress the original
# parse-failure log only when the recovery succeeds.
repaired = _escape_stray_ampersands(payload)
if repaired != payload:
try:
root = ET.fromstring(repaired) # noqa: S314 — eisy LAN traffic
except ET.ParseError:
_LOGGER.debug("WS frame XML parse failed (%s); frame=%r", exc, payload[:200])
return None
else:
_LOGGER.debug("WS frame XML parse failed (%s); frame=%r", exc, payload[:200])
return None
if root.tag != "Event":
return None
action_el = root.find("action")
uom, precision = _decode_action_attrs(action_el)
return Event(
seqnum=_int_or(root.get("seqnum", "0"), default=0),
timestamp=root.get("timestamp", ""),
control=_text(root.find("control")),
action=_text(action_el),
node_address=_text(root.find("node")),
formatted_action=root.findtext("fmtAct", default="") or "",
formatted_name=root.findtext("fmtName", default="") or "",
uom=uom,
precision=precision,
event_info=_extract_event_info(root),
)
def _extract_event_info(root: ET.Element) -> str:
"""Serialise ``<eventInfo>`` back to a string, or return ``""``.
Variable change frames pack ``<var type="..." id="..."><val>``,
network resource frames carry ``<eventInfo>`` plus typed children,
Z-Wave / Matter status frames carry config dicts, and controller
logs (``_7``) carry CDATA. The parser keeps the inner XML verbatim
so consumers can pick the parsing strategy that fits — most
consumers won't care, but when they do, re-parsing the frame
themselves would mean carrying the raw bytes alongside every
``Event``, defeating the value of having a parsed dataclass.
Empty ``<eventInfo/>`` and absent elements both return ``""``.
"""
info = root.find("eventInfo")
if info is None:
return ""
# Use a string builder over .text + child serialisation so that
# mixed-content nodes (CDATA + element children, like _7
# controller logs) round-trip without losing bits.
pieces: list[str] = []
if info.text:
pieces.append(info.text)
for child in info:
pieces.append(ET.tostring(child, encoding="unicode"))
if child.tail:
pieces.append(child.tail)
return "".join(pieces).strip()
def _text(element: ET.Element | None) -> str:
"""Read an element's text safely, treating absent elements as empty."""
if element is None:
return ""
return element.text or ""
def _int_or(raw: str, *, default: int) -> int:
"""Coerce a string to int; return ``default`` on failure."""
try:
return int(raw)
except ValueError:
return default
def _decode_action_attrs(action_el: ET.Element | None) -> tuple[str, int | None]:
"""Pull ``uom`` and ``prec`` attrs off an ``<action>`` element.
``prec`` is ``None`` when absent or non-numeric; legitimate negative
values (rare but possible per the IoX spec) round-trip unchanged.
"""
if action_el is None:
return "", None
uom = action_el.get("uom", "")
prec_raw = action_el.get("prec")
if prec_raw is None:
return uom, None
try:
return uom, int(prec_raw)
except ValueError:
return uom, None
#: Matches ``&`` that isn't already part of a valid XML/HTML entity
#: reference (``&`` / ``<`` / ``>`` / ``"`` / ``'``
#: / numeric ``&#NN;`` / hex ``&#xNN;``). Used to recover frames where
#: eisy embeds query-string-shaped text (``NUM.107=24&VAL.111=1``) in
#: ``<eventInfo>`` without escaping the ampersand.
_STRAY_AMPERSAND_RE = re.compile(r"&(?!(?:amp|lt|gt|quot|apos|#\d+|#x[0-9a-fA-F]+);)")
def _escape_stray_ampersands(payload: str) -> str:
"""Escape ``&`` chars that aren't part of an XML entity reference."""
return _STRAY_AMPERSAND_RE.sub("&", payload)
def _maybe_unwrap_json_envelope(raw: str) -> str | None:
"""Return the inner XML payload, or the raw string if unwrapped.
Returns ``None`` when the frame is a non-event JSON envelope
(e.g. ``"spolisy"`` PG3 status frames) or unparsable JSON that
also isn't XML-shaped — the dispatcher should ignore those.
"""
stripped = raw.lstrip()
if stripped.startswith("<"):
return raw
if not stripped.startswith("{"):
return None
try:
envelope = json.loads(raw)
except json.JSONDecodeError:
return None
if not isinstance(envelope, dict):
return None
if envelope.get("type") != "event":
# spolisy / null / unknown — not a property update.
return None
data = envelope.get("data")
return data if isinstance(data, str) else None
EventListener = Callable[[Event], None]
NodeLifecycleListener = Callable[[NodeLifecycleEvent], None]
[docs]
class ProgramRunState(IntEnum):
"""Low-nibble of the ``<s>`` byte on a program-status frame.
Cookbook §8.5.3: exactly one of three run-clause states per frame,
ORed with a :class:`ProgramEvalState` (high nibble) in the byte.
Absent (``None`` on the event) when the high nibble is
:attr:`ProgramEvalState.NOT_LOADED` — the program errored so
there's no clause currently running.
"""
IDLE = 0x01
THEN = 0x02
ELSE = 0x03
[docs]
class ProgramEvalState(IntEnum):
"""High-nibble of the ``<s>`` byte on a program-status frame.
Cookbook §8.5.3: the program's last if-clause evaluation result.
The ``status: bool`` field on :class:`ProgramStatusEvent` derives
from the same source (the ``<on/>``/``<off/>`` element) but this
enum disambiguates the three "not really True/False" cases that
the bool collapses.
.. note::
:attr:`NOT_LOADED` is the cookbook's literal label, but in
practice the controller emits ``0xF0`` when the program **failed
to compile or hit a runtime error** — not (only) when it hasn't
been loaded yet. Treat this as the program-error sentinel; see
:class:`ProgramRunState` for why ``run_state`` is ``None`` in
this case.
"""
UNKNOWN = 0x10
TRUE = 0x20
FALSE = 0x30
NOT_LOADED = 0xF0
def _extract_event_tz(timestamp: str) -> tzinfo | None:
"""Pull the controller's tz from a frame's ``<Event timestamp="...">``.
The eisy stamps every WS frame with its own local time + offset
(e.g. ``"2026-05-14T20:47:26.828098-05:00"``). That offset is the
most reliable signal pyisyox has for the controller's tz; the
naive ``YYMMDD HH:MM:SS`` body timestamps share it.
Returns ``None`` if the attribute is empty, unparsable, or carries
no offset — the caller falls back to system-local in that case.
"""
if not timestamp:
return None
try:
parsed = datetime.fromisoformat(timestamp)
except ValueError:
return None
return parsed.tzinfo
def _parse_ws_program_timestamp(raw: str | None, *, tz: tzinfo | None = None) -> str | None:
"""Convert a ``<r>`` / ``<f>`` / ``<nsr>`` WS timestamp to ISO 8601 UTC.
The eisy emits these as ``"YYMMDD HH:MM:SS "`` (note the trailing
space) in the **controller's local time**, with no timezone
indicator on the body element. Returns an ISO 8601 string with an
explicit UTC offset (e.g. ``"2026-05-14T21:44:11+00:00"``) so the
stored representation matches the REST ``Z``-suffix shape — the
typed :class:`pyisyox.runtime.Program` accessors then parse it
back to a tz-aware :class:`datetime` that consumers can hand
straight to HA's TIMESTAMP sensor class without a second tz
coercion.
Args:
raw: The wire string. ``None`` / blank / unparsable returns
``None`` so the dispatcher can treat "absent" and
"unparsable" the same way; the unparsable case is logged
at DEBUG to aid firmware-quirk triage.
tz: The controller's tz, normally extracted from the parent
frame's ``<Event timestamp="...">`` offset (see
:func:`_extract_event_tz`). When ``None`` the caller
doesn't know — fall back to system-local. The fall-back
is only correct when host tz == eisy tz; the fallback
path matters in test contexts where there's no ambient
event frame.
Note on the ``%y`` window: Python maps two-digit years 00-68 to
2000-2068 and 69-99 to 1969-1999. The eisy is a home controller
so timestamps past 2068 are not a realistic concern in this
decade, but worth flagging if the cookbook ever upgrades to
four-digit years.
"""
if raw is None:
return None
text = raw.strip()
if not text:
return None
try:
parsed = datetime.strptime(text, "%y%m%d %H:%M:%S")
except ValueError:
_LOGGER.debug("unparsable program WS timestamp %r — skipping", raw)
return None
# ``astimezone()`` on a naive datetime assumes system-local — the
# fall-back when no event frame supplies the eisy's tz (e.g. direct
# helper calls). The explicit-tz branch keeps the wall-clock when
# the frame did supply one.
if tz is not None: # noqa: SIM108
parsed = parsed.replace(tzinfo=tz)
else:
parsed = parsed.astimezone()
return parsed.astimezone(UTC).isoformat()
def _decode_program_status_byte(
raw: int | None,
) -> tuple[ProgramRunState | None, ProgramEvalState | None]:
"""Split the cookbook ``<s>`` byte into (run, eval) typed enums.
Each nibble's value is looked up against the matching enum;
unknown bit patterns yield ``None`` for that nibble rather than
raising, so a future firmware addition doesn't break the
dispatcher.
"""
if raw is None:
return (None, None)
try:
run_state: ProgramRunState | None = ProgramRunState(raw & 0x0F)
except ValueError:
run_state = None
try:
eval_state: ProgramEvalState | None = ProgramEvalState(raw & 0xF0)
except ValueError:
eval_state = None
return (run_state, eval_state)
[docs]
@dataclass(slots=True, frozen=True)
class ProgramStatusEvent:
"""A program toggled true/false on the controller.
Emitted by :class:`EventDispatcher` whenever a
``<control>_1</control>`` frame with ``<action>0</action>``
arrives carrying a program id in its ``<eventInfo>``. The
matching :class:`pyisyox.client.ProgramRecord` is mutated in
place before listeners fire, so consumers reading
``program.status`` from a callback see the updated value.
Attributes:
address: Program id (4-character hex, zero-padded to match
``/api/programs``).
status: ``True`` when the cookbook ``<s>`` byte's eval state
is :attr:`ProgramEvalState.TRUE` (the if-clause matched on
the most recent evaluation); ``False`` for
:attr:`ProgramEvalState.FALSE`. For
:attr:`ProgramEvalState.UNKNOWN` /
:attr:`ProgramEvalState.NOT_LOADED` (and frames with no
``<s>`` byte) the dispatcher carries forward the prior
``record.status`` so a transient unknown doesn't flip the
entity. Wire-shape note: the ``<on/>`` / ``<off/>``
elements that ride along on the same frame are the
**enabled flag**, not the status — see :attr:`enabled`.
running: Raw ``<s>`` byte the eisy sent, or ``None`` if
absent. Cookbook §8.5.3: the byte is a bitwise OR of a
:class:`ProgramRunState` (low nibble) and a
:class:`ProgramEvalState` (high nibble); use
``run_state`` / ``eval_state`` for the typed view.
run_state: Decoded low nibble — ``IDLE`` / ``THEN`` / ``ELSE``,
or ``None`` when the program isn't loaded
(``eval_state == NOT_LOADED``) or the wire byte was
absent / unrecognised.
eval_state: Decoded high nibble — ``UNKNOWN`` / ``TRUE`` /
``FALSE`` / ``NOT_LOADED``, or ``None`` when the wire
byte was absent / unrecognised. Disambiguates the
three "not really True/False" cases that ``status: bool``
collapses. ``NOT_LOADED`` is the cookbook label for what
is in practice the **program-errored** sentinel — see
:class:`ProgramEvalState`.
enabled: New ``enabled`` state when the frame carried an
``<on/>`` / ``<off/>`` element — ``True`` for ``<on/>``,
``False`` for ``<off/>``. ``None`` when the frame omitted
both (some "ran"-only frames carry only ``<r>`` / ``<f>``
/ ``<s>`` — see cookbook §8.5.3). The matching record's
:attr:`pyisyox.client.ProgramRecord.enabled` is updated
in-place before listeners fire when this is non-``None``.
run_at_startup: New ``run_at_startup`` state when the frame
carried an ``<rr/>`` (True) or ``<nr/>`` (False) element.
``None`` when the frame omitted both. Mirror of the
enabled-flag pattern; the record's
:attr:`pyisyox.client.ProgramRecord.run_at_startup` is
updated in-place before listeners fire.
seqnum: Sequence number of the underlying :class:`Event`.
"""
address: str
status: bool
running: int | None
seqnum: int
run_state: ProgramRunState | None = None
eval_state: ProgramEvalState | None = None
enabled: bool | None = None
run_at_startup: bool | None = None
ProgramStatusListener = Callable[[ProgramStatusEvent], None]
[docs]
@dataclass(slots=True, frozen=True)
class VariableTableChangeEvent:
"""A ``_1`` / action ``"9"`` system event — variable table changed.
The eisy fires this when a variable is added, removed, or has its
precision changed (a structural / metadata change that the
per-value :class:`TriggerAction.VARIABLE_VALUE` / ``VARIABLE_INIT``
frames don't carry). ``<eventInfo>`` payload::
<var><type>N</type><id>0</id></var>
``id=0`` is the wildcard sentinel — "this whole type's table
changed", not a specific variable. Consumers should re-fetch
``/api/variables/{type_id}`` to pick up the new metadata
(precision in particular — the per-value frames carry raw ``val``
and a stale precision will mis-render every variable of this
type until the registry is refreshed).
"""
#: Variable type — ``"1"`` (integer) or ``"2"`` (state).
type_id: str
seqnum: int
VariableTableChangeListener = Callable[[VariableTableChangeEvent], None]
def _parse_lifecycle_enabled(event_info: str) -> bool | None:
"""Pull the ``<enabled>`` flag off a ``NODE_ENABLED`` (``EN``) frame's
``<eventInfo>`` — ``EN`` covers both directions, the boolean says which.
Returns ``None`` when the flag is absent or unparsable (the caller
then leaves the record's ``enabled`` state untouched).
"""
if not event_info:
return None
try:
root = ET.fromstring(f"<eventInfo>{event_info}</eventInfo>") # noqa: S314 — eisy LAN traffic
except ET.ParseError:
return None
text = (root.findtext("enabled") or "").strip().lower()
if text in {"true", "1"}:
return True
if text in {"false", "0"}:
return False
return None
def _extract_lifecycle_node_xml(raw_frame: str) -> str | None:
"""Pull the inner ``<node>...</node>`` element text out of a lifecycle
frame's ``<eventInfo>``.
The eisy emits the full new node element on ``ND`` actions (capture
confirmed); other lifecycle verbs may follow the same pattern in
eventInfo but haven't been observed yet. Returns ``None`` when no
inner ``<node>`` is found — keeps the consumer code path simple.
"""
payload = _maybe_unwrap_json_envelope(raw_frame)
if payload is None:
return None
try:
root = ET.fromstring(payload) # noqa: S314 — eisy LAN traffic
except ET.ParseError:
return None
info = root.find("eventInfo")
if info is None:
return None
node_el = info.find("node")
if node_el is None:
return None
return ET.tostring(node_el, encoding="unicode")
[docs]
class EventDispatcher:
"""Routes parsed :class:`Event` instances into a node registry +
listener callbacks.
The dispatcher is intentionally not coupled to the WebSocket
transport — :meth:`feed` accepts a raw frame and does the parse +
route + emit dance. The actual WS read loop lives in
:mod:`pyisyox.runtime.ws`; tests can drive the dispatcher directly
with synthetic frames.
"""
__slots__ = (
"_group_members_index",
"_lifecycle_listeners",
"_listeners",
"_nodes",
"_program_status_listeners",
"_programs",
"_variable_table_change_listeners",
"_variables",
)
def __init__(
self,
nodes: dict[str, NodeRecord],
programs: dict[str, ProgramRecord] | None = None,
variables: dict[str, dict[str, VariableRecord]] | None = None,
groups: dict[str, GroupRecord] | None = None,
) -> None:
"""Bind to a node + program + variable + group registry.
The dispatcher mutates records in place. Events for unknown
addresses are dropped silently (subscribe to lifecycle for
joins). Passing ``None`` for ``programs``/``variables`` makes
those dispatch paths a no-op.
``groups`` restores pyisy-3.x parity: a group/scene carries no
wire status of its own (the controller never emits an event for
the group address), so a member node's property change is
re-emitted as a synthetic event addressed to each containing
group. Per-address subscribers on the group re-render and re-read
the (computed-on-access) :attr:`pyisyox.runtime.Group.group_any_on`.
Passing ``None`` disables the re-emit (legacy behaviour).
"""
self._nodes = nodes
self._programs = programs if programs is not None else {}
self._variables = variables if variables is not None else {}
self._group_members_index: dict[str, tuple[str, ...]] = {}
self.update_groups(groups if groups is not None else {})
self._listeners: list[EventListener] = []
self._lifecycle_listeners: list[NodeLifecycleListener] = []
self._program_status_listeners: list[ProgramStatusListener] = []
self._variable_table_change_listeners: list[VariableTableChangeListener] = []
[docs]
def update_groups(self, groups: dict[str, GroupRecord]) -> None:
"""(Re)build the member→groups reverse index from a group registry.
Called from ``__init__`` and again by
:meth:`pyisyox.controller.Controller.refresh` — ``refresh()``
replaces ``LoadResult.groups`` with a fresh dict (unlike
``nodes``, which is mutated in place), so the index has to be
rebuilt or scene-membership changes from a reload lifecycle
event would be missed (new members never re-emit; removed
members still would).
"""
members_index: dict[str, list[str]] = {}
for group_address, group_record in groups.items():
for member_address in group_record.member_addresses:
members_index.setdefault(member_address, []).append(group_address)
self._group_members_index = {
member: tuple(group_addresses) for member, group_addresses in members_index.items()
}
[docs]
def add_listener(self, callback: EventListener) -> Callable[[], None]:
"""Register ``callback`` to fire on every parsed event.
Returns:
An unsubscribe function. Calling it removes ``callback``
from the listener list. Safe to call from inside a
callback (the dispatcher iterates a snapshot).
"""
self._listeners.append(callback)
def _unsubscribe() -> None:
try:
self._listeners.remove(callback)
except ValueError:
pass
return _unsubscribe
[docs]
def add_program_status_listener(self, callback: ProgramStatusListener) -> Callable[[], None]:
"""Register ``callback`` to fire on every program-status frame
(``<control>_1</control>`` action ``"0"``).
The dispatcher updates the matching
:class:`pyisyox.client.ProgramRecord` in place before firing,
so consumers reading ``program.status`` from the callback see
the new value.
Returns:
An unsubscribe function.
"""
self._program_status_listeners.append(callback)
def _unsubscribe() -> None:
try:
self._program_status_listeners.remove(callback)
except ValueError:
pass
return _unsubscribe
[docs]
def add_variable_table_change_listener(self, callback: VariableTableChangeListener) -> Callable[[], None]:
"""Register ``callback`` to fire on every variable-table-change
frame (``<control>_1</control>`` action ``"9"``).
Fired when a variable is added, removed, or has its precision
changed on the controller. The dispatcher itself does **not**
re-fetch the variable table — the listener is the seam where
consumers wire in a focused re-fetch (e.g. by calling
:meth:`Controller.refresh`) so the registry mirrors the new
metadata. See :class:`VariableTableChangeEvent` for the payload.
Returns:
An unsubscribe function.
"""
self._variable_table_change_listeners.append(callback)
def _unsubscribe() -> None:
try:
self._variable_table_change_listeners.remove(callback)
except ValueError:
pass
return _unsubscribe
[docs]
def add_lifecycle_listener(self, callback: NodeLifecycleListener) -> Callable[[], None]:
"""Register ``callback`` to fire on every parsed
:class:`NodeLifecycleEvent` (``<control>_3</control>`` frames).
Use this to drive reload UX: HA Core typically registers a
Repair issue when it sees a lifecycle event with
``requires_reload=True``, prompting the user to reload the
integration when convenient. The dispatcher does **not**
update the node registry on lifecycle events — consumers
decide whether to call :meth:`pyisyox.controller.Controller.refresh`
or live with a stale view until manual reload.
Returns:
An unsubscribe function.
"""
self._lifecycle_listeners.append(callback)
def _unsubscribe() -> None:
try:
self._lifecycle_listeners.remove(callback)
except ValueError:
pass
return _unsubscribe
[docs]
def feed(self, raw_frame: str) -> Event | None:
"""Parse one frame, apply the property update, fan out to listeners.
Returns the parsed :class:`Event` for callers that want to
peek (e.g. for sequence-number tracking), or ``None`` when the
frame couldn't be parsed (malformed XML, non-event envelope,
keep-alive null). Never raises on bad input — a single bad
frame must not crash the read loop.
"""
event = parse_event_frame(raw_frame)
if event is None:
return None
# Routed frames (node property / lifecycle / program-status /
# variable-change) get a purpose-built DEBUG line from the
# handler that decodes them; anything else — heartbeats,
# system-status, the trigger sub-events we don't act on, the
# protocol-driver / billing / Matter / Z-Wave frames — gets the
# generic `_log_system_event` line. Property updates aren't
# logged here: consumers hold the entity/name mapping and log
# state changes in their own vocabulary.
if event.is_node_property:
self._apply_property_update(event)
# Lifecycle frames go through their own listener channel in
# addition to the general event channel — the typed
# NodeLifecycleEvent is more ergonomic for consumers driving
# reload UX than re-parsing the raw frame.
elif event.control == SystemEventControl.NODE_LIFECYCLE:
self._emit_lifecycle(event, raw_frame)
elif event.control == SystemEventControl.TRIGGER and event.action == TriggerAction.PROGRAM_STATUS:
self._apply_program_status(event)
elif event.control == SystemEventControl.TRIGGER and event.action in (
TriggerAction.VARIABLE_VALUE,
TriggerAction.VARIABLE_INIT,
):
self._apply_variable_change(event)
elif (
event.control == SystemEventControl.TRIGGER
and event.action == TriggerAction.VARIABLE_TABLE_CHANGED
):
self._apply_variable_table_change(event)
elif _LOGGER.isEnabledFor(logging.DEBUG):
_log_system_event(event)
for listener in tuple(self._listeners):
try:
listener(event)
except Exception: # pylint: disable=broad-except
_LOGGER.exception("event listener raised; suppressing to keep loop alive")
# pyisy-3.x parity: a member node's property change implies its
# containing scene(s) may have changed aggregate state. Groups
# have no wire event of their own, so synthesise one per group.
if event.is_node_property:
self._reemit_group_status(event)
return event
def _reemit_group_status(self, event: Event) -> None:
"""Re-publish a member property change as a group-addressed event.
Mirrors ``pyisy.nodes.Group`` re-emitting its own
``status_events`` when a member changed. The synthetic event is
a pure *notification*, not a status frame: it is **not** fed
back through :meth:`feed` (no re-parse, no
``_apply_property_update`` — groups aren't in the node
registry) and never recurses (groups can't be members of
groups). Per-address subscribers fire and re-read the
computed-on-access :attr:`pyisyox.runtime.Group.group_any_on`.
``action`` is deliberately **empty** (and ``uom`` /
``formatted_*`` left default): a group has no single status
value, and echoing the triggering member's raw value under
``control="ST"`` would be misleading to a consumer that reads
``event.action`` directly. ``control`` stays ``"ST"`` only so
the event routes on the group's normal per-address/status
channel; ``seqnum`` / ``timestamp`` are carried for ordering
and provenance.
"""
group_addresses = self._group_members_index.get(event.node_address)
if not group_addresses:
return
for group_address in group_addresses:
group_event = Event(
seqnum=event.seqnum,
timestamp=event.timestamp,
control=PROP_STATUS,
action="",
node_address=group_address,
)
for listener in tuple(self._listeners):
try:
listener(group_event)
except Exception: # pylint: disable=broad-except
_LOGGER.exception("group re-emit listener raised; suppressing to keep loop alive")
def _emit_lifecycle(self, event: Event, raw_frame: str) -> None:
"""Build a :class:`NodeLifecycleEvent` and fan to lifecycle listeners.
For ``NODE_ENABLED`` (``EN``) frames the record's ``enabled`` flag
is updated in place first — so a node (de)activated from the admin
console / REST is reflected through ``Node.enabled`` even though no
listener acted on it.
"""
enabled: bool | None = None
if event.action == NodeLifecycleAction.NODE_ENABLED:
enabled = _parse_lifecycle_enabled(event.event_info)
if enabled is not None:
record = self._nodes.get(event.node_address)
if record is not None:
record.enabled = enabled
if _LOGGER.isEnabledFor(logging.DEBUG):
extra = _compact_event_info(event.event_info)
_LOGGER.debug(
"Node lifecycle: %s on %s%s",
NodeLifecycleAction.label(event.action),
event.node_address or "(controller)",
f" {extra}" if extra else "",
)
if not self._lifecycle_listeners:
return
try:
action: NodeLifecycleAction | str = NodeLifecycleAction(event.action)
except ValueError:
action = event.action
node_xml = _extract_lifecycle_node_xml(raw_frame) if event.action == "ND" else None
lifecycle = NodeLifecycleEvent(
action=action,
node_address=event.node_address,
raw_action=event.action,
seqnum=event.seqnum,
node_xml=node_xml,
enabled=enabled,
)
for listener in tuple(self._lifecycle_listeners):
try:
listener(lifecycle)
except Exception: # pylint: disable=broad-except
_LOGGER.exception("lifecycle listener raised; suppressing to keep loop alive")
def _apply_property_update(self, event: Event) -> None:
"""Overlay an event's value into the matching node's properties."""
record = self._nodes.get(event.node_address)
if record is None:
_LOGGER.debug(
"WS event for unknown node address %r — dropping (control=%s)",
event.node_address,
event.control,
)
return
record.properties[event.control] = NodePropertyValue(
id=event.control,
value=event.action,
formatted=event.formatted_action,
uom=event.uom,
name=event.formatted_name,
precision=event.precision or 0,
)
if _LOGGER.isEnabledFor(logging.DEBUG):
display = event.formatted_action.strip() or event.action
extras = []
if display != event.action:
extras.append(f"raw={event.action}")
if event.uom:
extras.append(f"uom={event.uom}")
suffix = f" ({', '.join(extras)})" if extras else ""
_LOGGER.debug("Node %s %s -> %s%s", event.node_address, event.control, display, suffix)
def _apply_variable_change(self, event: Event) -> None:
"""Decode a variable-change frame and update the matching record.
Wire shape (same control as program-status, different action)::
<control>_1</control><action>6|7</action>
<eventInfo><var type="N" id="M"><val>123</val><ts>...</ts>...</var></eventInfo>
Action ``"6"`` is a current-value change; ``"7"`` is an init
(restore-on-startup) change. ``type`` is ``"1"`` (integer) or
``"2"`` (state); the ``VariableRecord`` registry is keyed
``{type_id: {id: record}}`` matching what
:func:`pyisyox.client.parse_api_variables_type` produces.
Unknown (type, id) pairs are dropped silently — typically a
variable created after the initial load. Consumers that care
can trigger a reload via :meth:`Controller.refresh`.
"""
if not event.event_info:
return
try:
info = ET.fromstring( # noqa: S314 — eisy LAN traffic
f"<eventInfo>{event.event_info}</eventInfo>"
)
except ET.ParseError:
return
var_elem = info.find("var")
if var_elem is None:
return
type_id = (var_elem.get("type") or "").strip()
var_id = (var_elem.get("id") or "").strip()
if not type_id or not var_id:
return
type_bucket = self._variables.get(type_id)
if type_bucket is None:
_LOGGER.debug("WS variable-change event for unknown type %r — dropping", type_id)
return
record = type_bucket.get(var_id)
if record is None:
_LOGGER.debug(
"WS variable-change event for unknown id %r/%r — dropping",
type_id,
var_id,
)
return
# Action 6 carries the new value in <val>; action 7 (init change)
# carries it in <init>. Old code read <val> regardless and
# silently dropped every init frame — fix is to pick the
# right element per action, with <val> as a tolerant fallback
# for action 7 in case some firmwares reuse the value field.
if event.action == TriggerAction.VARIABLE_INIT:
primary_text = (var_elem.findtext("init") or "").strip()
fallback_text = (var_elem.findtext("val") or "").strip()
field = "init"
else:
primary_text = (var_elem.findtext("val") or "").strip()
fallback_text = ""
field = "value"
text = primary_text or fallback_text
if not text:
return
try:
new_value: int | float = int(text)
except ValueError:
try:
new_value = float(text)
except ValueError:
_LOGGER.debug(
"WS variable-change event for %s.%s carried non-numeric value %r",
type_id,
var_id,
text,
)
return
if field == "value":
record.value = new_value
else:
record.init = new_value
ts_text = (var_elem.findtext("ts") or "").strip()
if ts_text:
record.ts = ts_text
_LOGGER.debug("Variable %s.%s %s updated to %s", type_id, var_id, field, new_value)
def _apply_variable_table_change(self, event: Event) -> None:
"""Decode a ``_1`` / action ``"9"`` variable-table-change frame
and fan out to ``add_variable_table_change_listener`` callbacks.
Wire shape::
<control>_1</control><action>9</action>
<eventInfo><var><type>N</type><id>0</id></var></eventInfo>
``id=0`` is the wildcard sentinel — the frame doesn't carry an
individual variable's new state, it just signals "this whole
type's table changed structurally (variable added / removed /
precision changed)". The dispatcher recognises the frame, logs
it, and emits a :class:`VariableTableChangeEvent` to registered
listeners. The dispatcher itself does **not** re-fetch — that's
the consumer's call (typically :meth:`Controller.refresh`).
Type is read from either an attribute (``<var type="2">``) or a
child element (``<var><type>2</type></var>``) — the wire shape
has been observed in both forms across firmwares.
"""
if not self._variable_table_change_listeners:
_LOGGER.debug("Variable table change (no listener) %s", event.event_info or "<empty>")
return
if not event.event_info:
return
try:
info = ET.fromstring( # noqa: S314 — eisy LAN traffic
f"<eventInfo>{event.event_info}</eventInfo>"
)
except ET.ParseError:
return
var_elem = info.find("var")
if var_elem is None:
return
type_id = (var_elem.get("type") or var_elem.findtext("type") or "").strip()
if not type_id:
return
evt = VariableTableChangeEvent(type_id=type_id, seqnum=event.seqnum)
_LOGGER.debug("Variable table change: type=%s seqnum=%s", type_id, event.seqnum)
for listener in tuple(self._variable_table_change_listeners):
try:
listener(evt)
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Variable table change listener raised")
def _apply_program_status(self, event: Event) -> None: # pylint: disable=too-many-locals
"""Decode a program-status frame and update the matching record.
Wire shape (cookbook §8.5.3)::
<control>_1</control><action>0</action>
<eventInfo>
<id>HEX</id>
<on/> or <off/> <!-- enabled flag -->
<rr/> or <nr/> <!-- run-at-reboot flag -->
<r>YYMMDD HH:MM:SS </r> <!-- last run start -->
<f>YYMMDD HH:MM:SS </f> <!-- last run finish -->
<nsr>YYMMDD HH:MM:SS </nsr> <!-- next scheduled run -->
<s>NN</s> <!-- status byte -->
</eventInfo>
Partial frames are common — e.g. just ``<id>`` + ``<nsr>``
when the scheduler plans the next run. The dispatcher mutates
only the fields the frame actually carries.
``<id>`` is hex without zero-padding (``8D``); the
``ProgramRecord`` registry is keyed on the zero-padded
4-character form (``008D``) from ``/api/programs``, so we
normalise.
Per the cookbook (and the pyisy 3.x reference): ``<on/>`` /
``<off/>`` are the **enabled** flag, *not* the status — they
mean "this program is enabled / disabled on the controller"
and ride on every program-status frame as the current state.
Status ("did the if-clause match") comes from the high nibble
of the ``<s>`` byte (:class:`ProgramEvalState`):
``TRUE`` → ``True``, ``FALSE`` → ``False``;
``UNKNOWN`` / ``NOT_LOADED`` carry forward the prior
``record.status`` rather than flipping based on a transient.
``<r>`` / ``<f>`` / ``<nsr>`` are timestamps in the
controller's local time as ``YYMMDD HH:MM:SS`` (with trailing
space): last-run start, last-finish, and next-scheduled-run
respectively. :func:`_parse_ws_program_timestamp` converts
them to UTC ISO 8601 strings on the record so the typed
:class:`pyisyox.runtime.Program` accessors decode either wire
shape (REST ``Z``-suffix UTC and WS-converted UTC) symmetrically.
``<nsr>`` often arrives in a *partial* frame on its own (just
``<id>`` + ``<nsr>``), e.g. immediately after the controller
plans the next run; the absent-field-preserves-prior-value
pattern keeps the rest of the record stable.
``<rr/>`` / ``<nr/>`` are the **run-at-reboot** flag (cookbook
§8.5.3 second flag, alongside enabled): ``<rr/>`` = on,
``<nr/>`` = off. (Confirmed against live captures from real
eisy hardware, 2026-05-14.) Note the naming overlap with
``<nsr>`` — the cookbook reuses ``nr`` for the boolean flag
rather than a "next run" timestamp.
"""
if not event.event_info:
return
try:
info = ET.fromstring( # noqa: S314 — eisy LAN traffic
f"<eventInfo>{event.event_info}</eventInfo>"
)
except ET.ParseError:
return
raw_id = (info.findtext("id") or "").strip()
if not raw_id:
return
# The wire id can be unpadded ('8D'); /api/programs zero-pads
# to 4 chars ('008D'). Try both so consumers see the update.
program_id = raw_id.zfill(4).upper()
record = self._programs.get(program_id) or self._programs.get(raw_id)
if record is None:
_LOGGER.debug("WS program-status event for unknown id %r — dropping", raw_id)
return
# <on/> / <off/> are the enabled flag (cookbook §8.5.3 +
# pyisy v3 ref). Absent on some frames; the dispatcher only
# touches record.enabled when the frame actually carries one.
# ``<onAdj/>`` / ``<offAdj/>`` are status-adjust markers that
# appear on *node* frames — pyisy v3 doesn't look for them on
# program-status frames, and no captured program-status frame
# carries one either. Intentionally not handled here; if a
# firmware ever emits them, the unknown-marker fall-through
# (``new_enabled = None``) keeps record.enabled unchanged
# rather than guessing.
new_enabled: bool | None
if info.find("on") is not None:
new_enabled = True
elif info.find("off") is not None:
new_enabled = False
else:
new_enabled = None
# ``<rr/>`` = run-at-reboot enabled, ``<nr/>`` = no run-at-
# reboot. Mutually exclusive on the wire (every captured frame
# carries exactly one). Same "absent → leave unchanged" pattern
# as the enabled flag.
new_run_at_startup: bool | None
if info.find("rr") is not None:
new_run_at_startup = True
elif info.find("nr") is not None:
new_run_at_startup = False
else:
new_run_at_startup = None
running_text = (info.findtext("s") or "").strip()
running_int: int | None
try:
# Cookbook §8.5.3: two ASCII hex digits.
running_int = int(running_text, 16) if running_text else None
except ValueError:
running_int = None
run_state, eval_state = _decode_program_status_byte(running_int)
# Status is the eval-state high nibble. UNKNOWN / NOT_LOADED /
# absent → carry forward the prior record.status (don't flip
# the entity on a transient mid-evaluation or a load error).
if eval_state is ProgramEvalState.TRUE:
new_status = True
elif eval_state is ProgramEvalState.FALSE:
new_status = False
else:
new_status = record.status
record.status = new_status
if new_enabled is not None:
record.enabled = new_enabled
if new_run_at_startup is not None:
record.run_at_startup = new_run_at_startup
if running_int is not None:
# Stored as the wire string verbatim so the typed Program
# accessors and consumers reading the raw byte agree on the
# source of truth; the typed ProgramStatusEvent carries the
# decoded int form.
record.running = running_text
# The controller stamps every WS frame with its own local
# time + offset (e.g. ``-05:00``). Use that as the tz for the
# naive ``YYMMDD HH:MM:SS`` body timestamps — more reliable
# than guessing the host's tz, which is wrong inside a
# ``TZ=UTC`` container with the eisy in a different zone.
event_tz = _extract_event_tz(event.timestamp)
last_run = _parse_ws_program_timestamp(info.findtext("r"), tz=event_tz)
last_finish = _parse_ws_program_timestamp(info.findtext("f"), tz=event_tz)
# ``<nsr>`` is the next-scheduled-run timestamp — fires as a
# standalone partial frame when the controller's scheduler
# plans the next run (e.g. immediately after ``runAtNextTime``
# is set or after each fire). Same wire shape as ``<r>``/``<f>``.
next_scheduled = _parse_ws_program_timestamp(info.findtext("nsr"), tz=event_tz)
if last_run is not None:
record.last_run_time = last_run
if last_finish is not None:
record.last_finish_time = last_finish
if next_scheduled is not None:
record.next_scheduled_run_time = next_scheduled
_LOGGER.debug(
"Program %s status -> %s%s%s%s",
record.address,
new_status,
f" enabled={new_enabled}" if new_enabled is not None else "",
f" run_at_startup={new_run_at_startup}" if new_run_at_startup is not None else "",
f" (running={running_int})" if running_int is not None else "",
)
if not self._program_status_listeners:
return
status_event = ProgramStatusEvent(
address=record.address,
status=new_status,
running=running_int,
seqnum=event.seqnum,
run_state=run_state,
eval_state=eval_state,
enabled=new_enabled,
run_at_startup=new_run_at_startup,
)
for listener in tuple(self._program_status_listeners):
try:
listener(status_event)
except Exception: # pylint: disable=broad-except
_LOGGER.exception("program-status listener raised; suppressing to keep loop alive")