Connectors

Talking to the device underneath.

py20305.connectors.base

Base connector class with async no-op defaults for all DER control modes.

Connectors inherit from BaseConnector and selectively override the methods they support. Unsupported modes are silent no-ops, matching the physical reality that not every device supports every IEEE 2030.5 control mode.

ConnectorConnectionError

Bases: ConnectorError

Raised when the connector cannot reach the target device.

The optional permanent flag signals that the failure won't change on retry without operator intervention -- e.g. Modbus protocol exception codes 1-4 (illegal function/address/value, server failure), which typically mean the configured device is missing models the connector was told to scan for.

Source code in src/py20305/connectors/errors.py

class ConnectorConnectionError(ConnectorError):
    """Raised when the connector cannot reach the target device.

    The optional ``permanent`` flag signals that the failure won't change
    on retry without operator intervention -- e.g. Modbus protocol
    exception codes 1-4 (illegal function/address/value, server failure),
    which typically mean the configured device is missing models the
    connector was told to scan for.
    """

    def __init__(self, *args: object, permanent: bool = False) -> None:
        super().__init__(*args)
        self.permanent = permanent

ConnectorError

Bases: Exception

Base exception for connector operations.

Source code in src/py20305/connectors/errors.py

class ConnectorError(Exception):
    """Base exception for connector operations."""

ConnectorTimeoutError

Bases: ConnectorError

Raised when a connector operation exceeds its deadline.

Source code in src/py20305/connectors/errors.py

class ConnectorTimeoutError(ConnectorError):
    """Raised when a connector operation exceeds its deadline."""

ConnectorValueError

Bases: ConnectorError

Raised when a control parameter is outside the range the profile allows.

Distinct from :class:ConnectorWriteError: nothing was attempted. The parameter itself is invalid, so no device could have accepted it and a retry with the same value cannot succeed. The register is left untouched, including any mode-enable point -- a lever must not be raised for a write that will not happen.

Reported to the head-end as IEEE 2030.5 Table 31 status 253 (Invalid), because the fault is in the event's data rather than in what the device can do. Adapters translate accordingly at their own boundary.

Source code in src/py20305/connectors/errors.py

class ConnectorValueError(ConnectorError):
    """Raised when a control parameter is outside the range the profile allows.

    Distinct from :class:`ConnectorWriteError`: nothing was attempted. The
    parameter itself is invalid, so no device could have accepted it and a retry
    with the same value cannot succeed. The register is left untouched, including
    any mode-enable point -- a lever must not be raised for a write that will not
    happen.

    Reported to the head-end as IEEE 2030.5 Table 31 status 253 (Invalid),
    because the fault is in the event's data rather than in what the device can
    do. Adapters translate accordingly at their own boundary.
    """

ConnectorWriteError

Bases: ConnectorError

Raised when a control write fails or is rejected by the device.

Source code in src/py20305/connectors/errors.py

class ConnectorWriteError(ConnectorError):
    """Raised when a control write fails or is rejected by the device."""

ScheduleNotification dataclass

A relayed, forward-looking schedule event for a connector.

Pushed by the core on every change to a relayed schedule -- a control, baseline, DOE or price -- so a downstream optimizer stays current without polling for it. Informational only: it does not apply setpoints, which remains the job of the update_* methods.

Source code in src/py20305/connectors/base.py

@dataclass(frozen=True)
class ScheduleNotification:
    """A relayed, forward-looking schedule event for a connector.

    Pushed by the core on every *change* to a relayed schedule -- a control,
    baseline, DOE or price -- so a downstream optimizer stays current without
    polling for it. Informational only: it does not apply setpoints, which
    remains the job of the ``update_*`` methods.
    """

    #: "control" | "default_baseline" | "doe" | "price".
    #: "drlc" | "flow_reservation" are reserved for later.
    stream: str
    #: event streams: "scheduled"|"updated"|"active"|"superseded"|"cancelled"|"completed";
    #: default_baseline: "default_added"|"default_updated".
    transition: str
    #: EventState value (scheduled/active/superseded/cancelled/completed); None for baseline.
    status: str | None
    #: Raw IEEE 2030.5-2018 §11.2.4 EventStatus.currentStatus (0-5); None for baseline.
    current_status: int | None
    #: DERControl / DefaultDERControl mRID, hex.
    mrid: str
    program_href: str
    #: Program primacy.
    primacy: int
    #: Effective (post-randomization) epoch seconds / seconds; None for baseline.
    start: int | None
    duration: int | None
    end: int | None
    #: Hex LFDIs this event scopes to, resolved by the core.
    affected_lfdis: list[str]
    #: (randomizeStart, randomizeDuration) seconds -- carried so optimizers time
    #: dispatch correctly (the update_* path drops it). None when unset.
    randomization: tuple[int | None, int | None]
    #: Stream-specific typed body (DERControlBase / DefaultDERControl / envelope limits).
    payload: dict[str, Any]

ReadingOverride dataclass

Per-quantity overrides for a connector's MirrorMeterReadings.

Returned (keyed by monitoring key, e.g. "W", "VL2") from :meth:BaseConnector.reading_overrides to tailor ReadingType metadata and the default qualityFlags to a connector's use case -- e.g. report a quantity as a Maximum (data_qualifier=8) or a Summation (accumulation_behaviour=9), or mark its readings derived.

Every field is optional; None leaves the client default in place, so a connector only states the deltas it cares about. The ReadingType fields map onto ReadingTypeSpec. quality_flags is a 16-bit value applied to the Reading (not the ReadingType). flow_direction is intentionally not overridable -- it stays derived from the value's sign.

Source code in src/py20305/connectors/base.py

@dataclass(frozen=True)
class ReadingOverride:
    """Per-quantity overrides for a connector's MirrorMeterReadings.

    Returned (keyed by monitoring key, e.g. ``"W"``, ``"VL2"``) from
    :meth:`BaseConnector.reading_overrides` to tailor ReadingType metadata and
    the default ``qualityFlags`` to a connector's use case -- e.g. report a
    quantity as a Maximum (``data_qualifier=8``) or a Summation
    (``accumulation_behaviour=9``), or mark its readings ``derived``.

    Every field is optional; ``None`` leaves the client default in place, so
    a connector only states the deltas it cares about. The ReadingType fields map
    onto ``ReadingTypeSpec``. ``quality_flags`` is a 16-bit value applied to the
    ``Reading`` (not the ReadingType). ``flow_direction`` is intentionally not
    overridable -- it stays derived from the value's sign.
    """

    accumulation_behaviour: int | None = None
    data_qualifier: int | None = None
    kind: int | None = None
    commodity: int | None = None
    multiplier: int | None = None  # powerOfTenMultiplier
    uom: int | None = None
    phase: int | None = None
    quality_flags: int | None = None

BaseConnector

Base connector with async no-op defaults for all DER methods.

Subclasses override only the methods their device supports. All methods are async to allow blocking I/O to be wrapped in executors.

Source code in src/py20305/connectors/base.py

class BaseConnector:
    """Base connector with async no-op defaults for all DER methods.

    Subclasses override only the methods their device supports.
    All methods are async to allow blocking I/O to be wrapped in executors.
    """

    connector_name: str = "BaseConnector"

    #: IEEE 2030.5 DER type code (see sep.Dertype).
    #: Common values: 4=PV, 81=EV, 82=EVSE, 83=Combined PV+Storage.
    der_type: int = 83

    #: What this connector speaks to its device, as reported in southbound
    #: telemetry. Defaults to "generic" — the consumer contract's catch-all —
    #: because the base class makes no claim about a wire: a connector that
    #: talks Modbus says so, and one that prints to a log should not be
    #: recorded as though it reached hardware.
    telemetry_protocol: str = "generic"

    # ------------------------------------------------------------------
    # Telemetry
    # ------------------------------------------------------------------

    async def fetch_monitoring(self) -> dict[str, Any]:
        """Return raw, unscaled power telemetry.

        Signs are referenced to the DER and are a requirement, not a hint:
        ``W`` SHALL be positive for generation and negative for absorption, and
        ``Var`` positive for reactive power injection (over-excited) and negative
        for absorption (under-excited). This is the SunSpec model 701 convention.
        The ``SIGNED_LOAD_CONVENTION`` reading profile negates both, so a
        connector reporting another convention posts the wrong sign.

        Beyond the keys below, a connector may supply per-line keys (``WL1``,
        ``VarL2``, ...), ``ACType`` (SunSpec 701 enum), phase-to-phase voltages
        (``VL1L2``, ``VL2L3``, ``VL3L1``) and State of Energy in Wh
        (``WHAvail``). Absent or ``None`` means not measured.
        """
        return {
            "W": None,
            "Var": None,
            "Hz": None,
            "V": None,
            "PF": None,
            "VA": None,
            "A": None,
        }

    def reading_overrides(self) -> dict[str, ReadingOverride] | None:
        """Optional per-quantity ReadingType / qualityFlags overrides.

        Keyed by the same monitoring key as ``fetch_monitoring`` (``"W"``,
        ``"VL2"``, ...). Return ``None`` (the default) to use the standard
        ReadingType metadata and ``qualityFlags=valid`` for every reading. A
        connector overrides only the deltas it needs; see :class:`ReadingOverride`.

        For a *static* per-key ``qualityFlags`` default, set it here. For a
        *per-cycle* value that varies with each sample (e.g. ``questionable`` on a
        stale read), put it in ``fetch_monitoring`` under ``"<key>__quality"``
        (e.g. ``{"W": 1500, "W__quality": 0x10}``); it takes precedence over the
        static default for that cycle.
        """
        return None

    async def fetch_nameplate(self) -> dict[str, Any]:
        """Return DER nameplate ratings."""
        return {
            "WMaxRtg": None,
            "WOvrExtRtg": None,
            "WOvrExtRtgPF": None,
            "WUndExtRtg": None,
            "WUndExtRtgPF": None,
            "VAMaxRtg": None,
            "VarMaxInjRtg": None,
            "VarMaxAbsRtg": None,
            "WChaRteMaxRtg": None,
            "VAChaRteMaxRtg": None,
            # Storage ratings (rtgMaxDischargeRateW/VA, rtgMaxWh, rtgMaxAh).
            # Omitted when None; a connector that reports them -- including an
            # explicit 0 -- has them serialized as-is.
            "WDisChaRteMaxRtg": None,
            "VADisChaRteMaxRtg": None,
            "WhMaxRtg": None,
            "AhMaxRtg": None,
            "VNomRtg": None,
            "VMaxRtg": None,
            "VMinRtg": None,
            "ReactSusceptRtg": None,
            "NorOpCatRtg": None,
            "AbnOpCatRtg": None,
            "CtrlModes": None,
            # CSIP-AUS only: DOE control bitmap (doeModesSupported). Connector
            # provides an int (4-bit bitmap); omitted/None defaults to all four
            # DOE limits (0x0F).
            "DoeModesSupported": None,
        }

    async def fetch_configuration(self) -> dict[str, Any]:
        """Return DER configuration settings."""
        return {
            "WMax": None,
            "WMaxOvrExt": None,
            "WOvrExtPF": None,
            "WMaxUndExt": None,
            "WUndExtPF": None,
            "VAMax": None,
            "VarMaxInj": None,
            "VarMaxAbs": None,
            "WChaRteMax": None,
            "VAChaRteMax": None,
            # Storage settings (setMaxDischargeRateW/VA, setMaxWh, setMaxAh).
            # Omitted when None; an explicit 0 is serialized as-is.
            "WDisChaRteMax": None,
            "VADisChaRteMax": None,
            "WhMax": None,
            "AhMax": None,
            "VNom": None,
            "VMax": None,
            "VMin": None,
            "CtrlModes": None,
            # CSIP-AUS only: enabled DOE control bitmap (doeModesEnabled).
            # Connector provides an int (4-bit bitmap); omitted/None -> export +
            # import limits (0x03).
            "DoeModesEnabled": None,
        }

    async def fetch_availability(self) -> dict[str, Any]:
        """Return DER availability / reserve status."""
        return {
            "availabilityDuration": None,
            "maxChargeDuration": None,
            "readingTime": None,
            "reserveChargePercent": None,
            "reservePercent": None,
            "statVarAvail": None,
            "statWAvail": None,
        }

    async def fetch_status(self) -> dict[str, Any]:
        """Return DER operational status."""
        return {
            "alarmStatus": 0,
            "connectStatus": {"dateTime": None, "value": None},
            "genConnectStatus": {"dateTime": None, "value": None},
            "inverterStatus": {"dateTime": None, "value": None},
            "manufacturerStatus": {"dateTime": None, "value": None},
            "operationalModeStatus": {"dateTime": None, "value": None},
            "stateOfChargeStatus": {"dateTime": None, "value": None},
            "localControlModeStatus": None,
            "storageModeStatus": None,
            "storConnectStatus": None,
            "readingTime": None,
        }

    # ------------------------------------------------------------------
    # Curve-based control modes
    # ------------------------------------------------------------------

    async def update_qv(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set Q(V) volt-var curve."""
        return None

    async def update_pv(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set P(V) volt-watt curve."""
        return None

    async def update_qp(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set Q(P) watt-var curve."""
        return None

    # ------------------------------------------------------------------
    # Power limits
    # ------------------------------------------------------------------

    async def update_p_lim(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set active power limit."""
        return None

    async def update_p_lim_inj(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set active power limit (injection)."""
        return None

    async def update_p_lim_abs(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set active power limit (absorption)."""
        return None

    # ------------------------------------------------------------------
    # Frequency response
    # ------------------------------------------------------------------

    async def update_freq_watt(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set curve-based frequency-watt mode."""
        return None

    async def update_pf(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set frequency droop parameters."""
        return None

    # ------------------------------------------------------------------
    # Reactive power
    # ------------------------------------------------------------------

    async def update_const_q(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set constant reactive power."""
        return None

    async def update_const_pf(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set constant power factor."""
        return None

    # ------------------------------------------------------------------
    # Voltage ride-through
    # ------------------------------------------------------------------

    async def update_ov(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set over-voltage ride-through (must trip)."""
        return None

    async def update_uv(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set under-voltage ride-through (must trip)."""
        return None

    async def update_ov_mc(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set over-voltage momentary cessation."""
        return None

    async def update_uv_mc(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set under-voltage momentary cessation."""
        return None

    # ------------------------------------------------------------------
    # Frequency ride-through
    # ------------------------------------------------------------------

    async def update_of(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set over-frequency ride-through (must trip)."""
        return None

    async def update_uf(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set under-frequency ride-through (must trip)."""
        return None

    # ------------------------------------------------------------------
    # Fixed output and ramp
    # ------------------------------------------------------------------

    async def update_fixed_w(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set fixed active power output."""
        return None

    async def update_p_ramp(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set active power ramp rate."""
        return None

    async def update_es_permit_service(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set enter service permission."""
        return None

    # ------------------------------------------------------------------
    # IEEE 2030.5-2023 new modes (Gaps 16-19)
    # ------------------------------------------------------------------

    async def update_connect(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set connect/energize state (AND logic)."""
        return None

    async def update_max_lim_pct_va_absorb(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set max VA absorption as percent."""
        return None

    async def update_max_lim_pct_va_inject(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set max VA injection as percent."""
        return None

    async def update_max_lim_pct_var_absorb(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set max var absorption as percent."""
        return None

    async def update_max_lim_pct_var_inject(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set max var injection as percent."""
        return None

    async def update_max_lim_pct_w_absorb(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set max W absorption as percent."""
        return None

    async def update_max_lim_var_absorb(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set max var absorption (absolute)."""
        return None

    async def update_max_lim_var_inject(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set max var injection (absolute)."""
        return None

    async def update_target_v(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set target voltage output."""
        return None

    async def update_target_var(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set target reactive power output."""
        return None

    async def update_target_w(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set target active power output."""
        return None

    async def update_delta_w(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set delta active power change."""
        return None

    async def update_delta_var(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set delta reactive power change."""
        return None

    async def update_fixed_v(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set fixed voltage setpoint."""
        return None

    async def update_grid_connect_permit(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set grid connect permission."""
        return None

    async def update_island_permit(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set island operation permission."""
        return None

    # ------------------------------------------------------------------
    # CSIP-AUS extension modes
    # ------------------------------------------------------------------

    async def update_exp_lim(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set export power limit."""
        return None

    async def update_imp_lim(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set import power limit."""
        return None

    async def update_gen_lim(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set generation power limit."""
        return None

    async def update_load_lim(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set load power limit."""
        return None

    # ------------------------------------------------------------------
    # Pricing
    # ------------------------------------------------------------------

    async def update_pricing_mode(self, params: dict[str, Any]) -> ConnectorPayload:
        """Set pricing mode data."""
        return None

    # ------------------------------------------------------------------
    # Schedule notifications (informational push; see ScheduleNotification)
    # ------------------------------------------------------------------

    async def on_schedule_notification(self, notification: ScheduleNotification) -> None:
        """Single transport for all relayed schedule streams.

        Called by the core on every change to a control/baseline affecting this
        connector, so downstream logic has current schedule state without
        polling. The default fans out to the per-stream ``notification_*``
        convenience methods; override either this transport or the specific
        stream methods. Informational only -- exceptions are logged and
        swallowed by the core, never affecting dispatch.
        """
        handler = getattr(self, f"notification_{notification.stream}", None)
        if handler is not None:
            await handler(notification)

    async def notification_control(self, notification: ScheduleNotification) -> None:
        """A scheduled DERControl was added/updated/activated/superseded/etc."""
        return None

    async def notification_default_baseline(self, notification: ScheduleNotification) -> None:
        """The program's DefaultDERControl baseline was added or changed."""
        return None

    async def notification_doe(self, notification: ScheduleNotification) -> None:
        """A Dynamic Operating Envelope (export/import/gen/load W-limit) changed."""
        return None

    async def notification_price(self, notification: ScheduleNotification) -> None:
        """The active pricing interval (TimeTariffInterval) changed. Relay-only:
        the connector decides whether to act on the price."""
        return None

    async def notification_drlc(self, notification: ScheduleNotification) -> None:
        """A demand-response (EndDeviceControl) schedule changed (future stream)."""
        return None

    async def notification_flow_reservation(self, notification: ScheduleNotification) -> None:
        """A flow-reservation window changed (future stream)."""
        return None

fetch_monitoring async

fetch_monitoring() -> dict[str, Any]

Return raw, unscaled power telemetry.

Signs are referenced to the DER and are a requirement, not a hint: W SHALL be positive for generation and negative for absorption, and Var positive for reactive power injection (over-excited) and negative for absorption (under-excited). This is the SunSpec model 701 convention. The SIGNED_LOAD_CONVENTION reading profile negates both, so a connector reporting another convention posts the wrong sign.

Beyond the keys below, a connector may supply per-line keys (WL1, VarL2, ...), ACType (SunSpec 701 enum), phase-to-phase voltages (VL1L2, VL2L3, VL3L1) and State of Energy in Wh (WHAvail). Absent or None means not measured.

Source code in src/py20305/connectors/base.py

async def fetch_monitoring(self) -> dict[str, Any]:
    """Return raw, unscaled power telemetry.

    Signs are referenced to the DER and are a requirement, not a hint:
    ``W`` SHALL be positive for generation and negative for absorption, and
    ``Var`` positive for reactive power injection (over-excited) and negative
    for absorption (under-excited). This is the SunSpec model 701 convention.
    The ``SIGNED_LOAD_CONVENTION`` reading profile negates both, so a
    connector reporting another convention posts the wrong sign.

    Beyond the keys below, a connector may supply per-line keys (``WL1``,
    ``VarL2``, ...), ``ACType`` (SunSpec 701 enum), phase-to-phase voltages
    (``VL1L2``, ``VL2L3``, ``VL3L1``) and State of Energy in Wh
    (``WHAvail``). Absent or ``None`` means not measured.
    """
    return {
        "W": None,
        "Var": None,
        "Hz": None,
        "V": None,
        "PF": None,
        "VA": None,
        "A": None,
    }

reading_overrides

reading_overrides() -> dict[str, ReadingOverride] | None

Optional per-quantity ReadingType / qualityFlags overrides.

Keyed by the same monitoring key as fetch_monitoring ("W", "VL2", ...). Return None (the default) to use the standard ReadingType metadata and qualityFlags=valid for every reading. A connector overrides only the deltas it needs; see :class:ReadingOverride.

For a static per-key qualityFlags default, set it here. For a per-cycle value that varies with each sample (e.g. questionable on a stale read), put it in fetch_monitoring under "<key>__quality" (e.g. {"W": 1500, "W__quality": 0x10}); it takes precedence over the static default for that cycle.

Source code in src/py20305/connectors/base.py

def reading_overrides(self) -> dict[str, ReadingOverride] | None:
    """Optional per-quantity ReadingType / qualityFlags overrides.

    Keyed by the same monitoring key as ``fetch_monitoring`` (``"W"``,
    ``"VL2"``, ...). Return ``None`` (the default) to use the standard
    ReadingType metadata and ``qualityFlags=valid`` for every reading. A
    connector overrides only the deltas it needs; see :class:`ReadingOverride`.

    For a *static* per-key ``qualityFlags`` default, set it here. For a
    *per-cycle* value that varies with each sample (e.g. ``questionable`` on a
    stale read), put it in ``fetch_monitoring`` under ``"<key>__quality"``
    (e.g. ``{"W": 1500, "W__quality": 0x10}``); it takes precedence over the
    static default for that cycle.
    """
    return None

fetch_nameplate async

fetch_nameplate() -> dict[str, Any]

Return DER nameplate ratings.

Source code in src/py20305/connectors/base.py

async def fetch_nameplate(self) -> dict[str, Any]:
    """Return DER nameplate ratings."""
    return {
        "WMaxRtg": None,
        "WOvrExtRtg": None,
        "WOvrExtRtgPF": None,
        "WUndExtRtg": None,
        "WUndExtRtgPF": None,
        "VAMaxRtg": None,
        "VarMaxInjRtg": None,
        "VarMaxAbsRtg": None,
        "WChaRteMaxRtg": None,
        "VAChaRteMaxRtg": None,
        # Storage ratings (rtgMaxDischargeRateW/VA, rtgMaxWh, rtgMaxAh).
        # Omitted when None; a connector that reports them -- including an
        # explicit 0 -- has them serialized as-is.
        "WDisChaRteMaxRtg": None,
        "VADisChaRteMaxRtg": None,
        "WhMaxRtg": None,
        "AhMaxRtg": None,
        "VNomRtg": None,
        "VMaxRtg": None,
        "VMinRtg": None,
        "ReactSusceptRtg": None,
        "NorOpCatRtg": None,
        "AbnOpCatRtg": None,
        "CtrlModes": None,
        # CSIP-AUS only: DOE control bitmap (doeModesSupported). Connector
        # provides an int (4-bit bitmap); omitted/None defaults to all four
        # DOE limits (0x0F).
        "DoeModesSupported": None,
    }

fetch_configuration async

fetch_configuration() -> dict[str, Any]

Return DER configuration settings.

Source code in src/py20305/connectors/base.py

async def fetch_configuration(self) -> dict[str, Any]:
    """Return DER configuration settings."""
    return {
        "WMax": None,
        "WMaxOvrExt": None,
        "WOvrExtPF": None,
        "WMaxUndExt": None,
        "WUndExtPF": None,
        "VAMax": None,
        "VarMaxInj": None,
        "VarMaxAbs": None,
        "WChaRteMax": None,
        "VAChaRteMax": None,
        # Storage settings (setMaxDischargeRateW/VA, setMaxWh, setMaxAh).
        # Omitted when None; an explicit 0 is serialized as-is.
        "WDisChaRteMax": None,
        "VADisChaRteMax": None,
        "WhMax": None,
        "AhMax": None,
        "VNom": None,
        "VMax": None,
        "VMin": None,
        "CtrlModes": None,
        # CSIP-AUS only: enabled DOE control bitmap (doeModesEnabled).
        # Connector provides an int (4-bit bitmap); omitted/None -> export +
        # import limits (0x03).
        "DoeModesEnabled": None,
    }

fetch_availability async

fetch_availability() -> dict[str, Any]

Return DER availability / reserve status.

Source code in src/py20305/connectors/base.py

async def fetch_availability(self) -> dict[str, Any]:
    """Return DER availability / reserve status."""
    return {
        "availabilityDuration": None,
        "maxChargeDuration": None,
        "readingTime": None,
        "reserveChargePercent": None,
        "reservePercent": None,
        "statVarAvail": None,
        "statWAvail": None,
    }

fetch_status async

fetch_status() -> dict[str, Any]

Return DER operational status.

Source code in src/py20305/connectors/base.py

async def fetch_status(self) -> dict[str, Any]:
    """Return DER operational status."""
    return {
        "alarmStatus": 0,
        "connectStatus": {"dateTime": None, "value": None},
        "genConnectStatus": {"dateTime": None, "value": None},
        "inverterStatus": {"dateTime": None, "value": None},
        "manufacturerStatus": {"dateTime": None, "value": None},
        "operationalModeStatus": {"dateTime": None, "value": None},
        "stateOfChargeStatus": {"dateTime": None, "value": None},
        "localControlModeStatus": None,
        "storageModeStatus": None,
        "storConnectStatus": None,
        "readingTime": None,
    }

update_qv async

update_qv(params: dict[str, Any]) -> ConnectorPayload

Set Q(V) volt-var curve.

Source code in src/py20305/connectors/base.py

async def update_qv(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set Q(V) volt-var curve."""
    return None

update_pv async

update_pv(params: dict[str, Any]) -> ConnectorPayload

Set P(V) volt-watt curve.

Source code in src/py20305/connectors/base.py

async def update_pv(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set P(V) volt-watt curve."""
    return None

update_qp async

update_qp(params: dict[str, Any]) -> ConnectorPayload

Set Q(P) watt-var curve.

Source code in src/py20305/connectors/base.py

async def update_qp(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set Q(P) watt-var curve."""
    return None

update_p_lim async

update_p_lim(params: dict[str, Any]) -> ConnectorPayload

Set active power limit.

Source code in src/py20305/connectors/base.py

async def update_p_lim(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set active power limit."""
    return None

update_p_lim_inj async

update_p_lim_inj(params: dict[str, Any]) -> ConnectorPayload

Set active power limit (injection).

Source code in src/py20305/connectors/base.py

async def update_p_lim_inj(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set active power limit (injection)."""
    return None

update_p_lim_abs async

update_p_lim_abs(params: dict[str, Any]) -> ConnectorPayload

Set active power limit (absorption).

Source code in src/py20305/connectors/base.py

async def update_p_lim_abs(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set active power limit (absorption)."""
    return None

update_freq_watt async

update_freq_watt(params: dict[str, Any]) -> ConnectorPayload

Set curve-based frequency-watt mode.

Source code in src/py20305/connectors/base.py

async def update_freq_watt(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set curve-based frequency-watt mode."""
    return None

update_pf async

update_pf(params: dict[str, Any]) -> ConnectorPayload

Set frequency droop parameters.

Source code in src/py20305/connectors/base.py

async def update_pf(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set frequency droop parameters."""
    return None

update_const_q async

update_const_q(params: dict[str, Any]) -> ConnectorPayload

Set constant reactive power.

Source code in src/py20305/connectors/base.py

async def update_const_q(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set constant reactive power."""
    return None

update_const_pf async

update_const_pf(params: dict[str, Any]) -> ConnectorPayload

Set constant power factor.

Source code in src/py20305/connectors/base.py

async def update_const_pf(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set constant power factor."""
    return None

update_ov async

update_ov(params: dict[str, Any]) -> ConnectorPayload

Set over-voltage ride-through (must trip).

Source code in src/py20305/connectors/base.py

async def update_ov(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set over-voltage ride-through (must trip)."""
    return None

update_uv async

update_uv(params: dict[str, Any]) -> ConnectorPayload

Set under-voltage ride-through (must trip).

Source code in src/py20305/connectors/base.py

async def update_uv(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set under-voltage ride-through (must trip)."""
    return None

update_ov_mc async

update_ov_mc(params: dict[str, Any]) -> ConnectorPayload

Set over-voltage momentary cessation.

Source code in src/py20305/connectors/base.py

async def update_ov_mc(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set over-voltage momentary cessation."""
    return None

update_uv_mc async

update_uv_mc(params: dict[str, Any]) -> ConnectorPayload

Set under-voltage momentary cessation.

Source code in src/py20305/connectors/base.py

async def update_uv_mc(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set under-voltage momentary cessation."""
    return None

update_of async

update_of(params: dict[str, Any]) -> ConnectorPayload

Set over-frequency ride-through (must trip).

Source code in src/py20305/connectors/base.py

async def update_of(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set over-frequency ride-through (must trip)."""
    return None

update_uf async

update_uf(params: dict[str, Any]) -> ConnectorPayload

Set under-frequency ride-through (must trip).

Source code in src/py20305/connectors/base.py

async def update_uf(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set under-frequency ride-through (must trip)."""
    return None

update_fixed_w async

update_fixed_w(params: dict[str, Any]) -> ConnectorPayload

Set fixed active power output.

Source code in src/py20305/connectors/base.py

async def update_fixed_w(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set fixed active power output."""
    return None

update_p_ramp async

update_p_ramp(params: dict[str, Any]) -> ConnectorPayload

Set active power ramp rate.

Source code in src/py20305/connectors/base.py

async def update_p_ramp(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set active power ramp rate."""
    return None

update_es_permit_service async

update_es_permit_service(params: dict[str, Any]) -> ConnectorPayload

Set enter service permission.

Source code in src/py20305/connectors/base.py

async def update_es_permit_service(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set enter service permission."""
    return None

update_connect async

update_connect(params: dict[str, Any]) -> ConnectorPayload

Set connect/energize state (AND logic).

Source code in src/py20305/connectors/base.py

async def update_connect(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set connect/energize state (AND logic)."""
    return None

update_max_lim_pct_va_absorb async

update_max_lim_pct_va_absorb(params: dict[str, Any]) -> ConnectorPayload

Set max VA absorption as percent.

Source code in src/py20305/connectors/base.py

async def update_max_lim_pct_va_absorb(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set max VA absorption as percent."""
    return None

update_max_lim_pct_va_inject async

update_max_lim_pct_va_inject(params: dict[str, Any]) -> ConnectorPayload

Set max VA injection as percent.

Source code in src/py20305/connectors/base.py

async def update_max_lim_pct_va_inject(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set max VA injection as percent."""
    return None

update_max_lim_pct_var_absorb async

update_max_lim_pct_var_absorb(params: dict[str, Any]) -> ConnectorPayload

Set max var absorption as percent.

Source code in src/py20305/connectors/base.py

async def update_max_lim_pct_var_absorb(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set max var absorption as percent."""
    return None

update_max_lim_pct_var_inject async

update_max_lim_pct_var_inject(params: dict[str, Any]) -> ConnectorPayload

Set max var injection as percent.

Source code in src/py20305/connectors/base.py

async def update_max_lim_pct_var_inject(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set max var injection as percent."""
    return None

update_max_lim_pct_w_absorb async

update_max_lim_pct_w_absorb(params: dict[str, Any]) -> ConnectorPayload

Set max W absorption as percent.

Source code in src/py20305/connectors/base.py

async def update_max_lim_pct_w_absorb(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set max W absorption as percent."""
    return None

update_max_lim_var_absorb async

update_max_lim_var_absorb(params: dict[str, Any]) -> ConnectorPayload

Set max var absorption (absolute).

Source code in src/py20305/connectors/base.py

async def update_max_lim_var_absorb(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set max var absorption (absolute)."""
    return None

update_max_lim_var_inject async

update_max_lim_var_inject(params: dict[str, Any]) -> ConnectorPayload

Set max var injection (absolute).

Source code in src/py20305/connectors/base.py

async def update_max_lim_var_inject(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set max var injection (absolute)."""
    return None

update_target_v async

update_target_v(params: dict[str, Any]) -> ConnectorPayload

Set target voltage output.

Source code in src/py20305/connectors/base.py

async def update_target_v(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set target voltage output."""
    return None

update_target_var async

update_target_var(params: dict[str, Any]) -> ConnectorPayload

Set target reactive power output.

Source code in src/py20305/connectors/base.py

async def update_target_var(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set target reactive power output."""
    return None

update_target_w async

update_target_w(params: dict[str, Any]) -> ConnectorPayload

Set target active power output.

Source code in src/py20305/connectors/base.py

async def update_target_w(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set target active power output."""
    return None

update_delta_w async

update_delta_w(params: dict[str, Any]) -> ConnectorPayload

Set delta active power change.

Source code in src/py20305/connectors/base.py

async def update_delta_w(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set delta active power change."""
    return None

update_delta_var async

update_delta_var(params: dict[str, Any]) -> ConnectorPayload

Set delta reactive power change.

Source code in src/py20305/connectors/base.py

async def update_delta_var(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set delta reactive power change."""
    return None

update_fixed_v async

update_fixed_v(params: dict[str, Any]) -> ConnectorPayload

Set fixed voltage setpoint.

Source code in src/py20305/connectors/base.py

async def update_fixed_v(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set fixed voltage setpoint."""
    return None

update_grid_connect_permit async

update_grid_connect_permit(params: dict[str, Any]) -> ConnectorPayload

Set grid connect permission.

Source code in src/py20305/connectors/base.py

async def update_grid_connect_permit(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set grid connect permission."""
    return None

update_island_permit async

update_island_permit(params: dict[str, Any]) -> ConnectorPayload

Set island operation permission.

Source code in src/py20305/connectors/base.py

async def update_island_permit(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set island operation permission."""
    return None

update_exp_lim async

update_exp_lim(params: dict[str, Any]) -> ConnectorPayload

Set export power limit.

Source code in src/py20305/connectors/base.py

async def update_exp_lim(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set export power limit."""
    return None

update_imp_lim async

update_imp_lim(params: dict[str, Any]) -> ConnectorPayload

Set import power limit.

Source code in src/py20305/connectors/base.py

async def update_imp_lim(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set import power limit."""
    return None

update_gen_lim async

update_gen_lim(params: dict[str, Any]) -> ConnectorPayload

Set generation power limit.

Source code in src/py20305/connectors/base.py

async def update_gen_lim(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set generation power limit."""
    return None

update_load_lim async

update_load_lim(params: dict[str, Any]) -> ConnectorPayload

Set load power limit.

Source code in src/py20305/connectors/base.py

async def update_load_lim(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set load power limit."""
    return None

update_pricing_mode async

update_pricing_mode(params: dict[str, Any]) -> ConnectorPayload

Set pricing mode data.

Source code in src/py20305/connectors/base.py

async def update_pricing_mode(self, params: dict[str, Any]) -> ConnectorPayload:
    """Set pricing mode data."""
    return None

on_schedule_notification async

on_schedule_notification(notification: ScheduleNotification) -> None

Single transport for all relayed schedule streams.

Called by the core on every change to a control/baseline affecting this connector, so downstream logic has current schedule state without polling. The default fans out to the per-stream notification_* convenience methods; override either this transport or the specific stream methods. Informational only -- exceptions are logged and swallowed by the core, never affecting dispatch.

Source code in src/py20305/connectors/base.py

async def on_schedule_notification(self, notification: ScheduleNotification) -> None:
    """Single transport for all relayed schedule streams.

    Called by the core on every change to a control/baseline affecting this
    connector, so downstream logic has current schedule state without
    polling. The default fans out to the per-stream ``notification_*``
    convenience methods; override either this transport or the specific
    stream methods. Informational only -- exceptions are logged and
    swallowed by the core, never affecting dispatch.
    """
    handler = getattr(self, f"notification_{notification.stream}", None)
    if handler is not None:
        await handler(notification)

notification_control async

notification_control(notification: ScheduleNotification) -> None

A scheduled DERControl was added/updated/activated/superseded/etc.

Source code in src/py20305/connectors/base.py

async def notification_control(self, notification: ScheduleNotification) -> None:
    """A scheduled DERControl was added/updated/activated/superseded/etc."""
    return None

notification_default_baseline async

notification_default_baseline(notification: ScheduleNotification) -> None

The program's DefaultDERControl baseline was added or changed.

Source code in src/py20305/connectors/base.py

async def notification_default_baseline(self, notification: ScheduleNotification) -> None:
    """The program's DefaultDERControl baseline was added or changed."""
    return None

notification_doe async

notification_doe(notification: ScheduleNotification) -> None

A Dynamic Operating Envelope (export/import/gen/load W-limit) changed.

Source code in src/py20305/connectors/base.py

async def notification_doe(self, notification: ScheduleNotification) -> None:
    """A Dynamic Operating Envelope (export/import/gen/load W-limit) changed."""
    return None

notification_price async

notification_price(notification: ScheduleNotification) -> None

The active pricing interval (TimeTariffInterval) changed. Relay-only: the connector decides whether to act on the price.

Source code in src/py20305/connectors/base.py

async def notification_price(self, notification: ScheduleNotification) -> None:
    """The active pricing interval (TimeTariffInterval) changed. Relay-only:
    the connector decides whether to act on the price."""
    return None

notification_drlc async

notification_drlc(notification: ScheduleNotification) -> None

A demand-response (EndDeviceControl) schedule changed (future stream).

Source code in src/py20305/connectors/base.py

async def notification_drlc(self, notification: ScheduleNotification) -> None:
    """A demand-response (EndDeviceControl) schedule changed (future stream)."""
    return None

notification_flow_reservation async

notification_flow_reservation(notification: ScheduleNotification) -> None

A flow-reservation window changed (future stream).

Source code in src/py20305/connectors/base.py

async def notification_flow_reservation(self, notification: ScheduleNotification) -> None:
    """A flow-reservation window changed (future stream)."""
    return None

py20305.connectors.config

Typed configuration for the devices a connector talks to.

One model per connector type, discriminated on type, so a configuration file naming a connector gets that connector's fields validated and nothing else. :data:DeviceConfig is the union to annotate with; :class:~py20305.connectors.registry.ConnectorConfigRegistry consumes it.

To add a connector type, subclass :class:_DeviceConfigBase with a literal type and add it to the union below and to the registry's class map.

SunSpecDeviceConfig

Bases: _DeviceConfigBase

SunSpec Modbus device configuration.

Covers the three transports the connector speaks: plain Modbus TCP, Modbus RTU over a serial port, and Modbus TCP wrapped in TLS. Fields for the transports not in use are ignored, so one shape covers all three.

Source code in src/py20305/connectors/config.py

class SunSpecDeviceConfig(_DeviceConfigBase):
    """SunSpec Modbus device configuration.

    Covers the three transports the connector speaks: plain Modbus TCP,
    Modbus RTU over a serial port, and Modbus TCP wrapped in TLS. Fields for
    the transports not in use are ignored, so one shape covers all three.
    """

    type: Literal["sunspec"] = "sunspec"
    transport: Literal["tcp", "rtu", "tcp+tls"] = "tcp"
    host: str = "127.0.0.1"
    port: int = 8502
    serial_port: str = "/dev/ttyUSB0"
    baudrate: int = 9600
    parity: str = "N"
    ca_path: str | None = None
    cert_path: str | None = None
    key_path: str | None = None
    insecure: bool = False
    unit_id: int = 1
    timeout: int = 5
    scan_retries: int = 3
    scan_retry_delay: float = 2.0
    der_type: int = 83

PrintDemoDeviceConfig

Bases: _DeviceConfigBase

A device that logs what it would have done instead of doing it.

Needs no hardware, which makes it the connector to point at when trying the client against a utility server for the first time.

Source code in src/py20305/connectors/config.py

class PrintDemoDeviceConfig(_DeviceConfigBase):
    """A device that logs what it would have done instead of doing it.

    Needs no hardware, which makes it the connector to point at when trying
    the client against a utility server for the first time.
    """

    type: Literal["print_demo"] = "print_demo"

CustomDeviceConfig

Bases: _DeviceConfigBase

A connector supplied by import path.

The escape hatch for a device this package has no connector for: point class_path at your own :class:~py20305.connectors.base.BaseConnector subclass and it is constructed with init_kwargs.

Source code in src/py20305/connectors/config.py

class CustomDeviceConfig(_DeviceConfigBase):
    """A connector supplied by import path.

    The escape hatch for a device this package has no connector for: point
    ``class_path`` at your own
    :class:`~py20305.connectors.base.BaseConnector` subclass and it
    is constructed with ``init_kwargs``.
    """

    type: Literal["custom"] = "custom"
    class_path: str = Field(description="Full Python import path for connector class")
    init_kwargs: dict[str, Any] = Field(default_factory=dict)

py20305.connectors.registry

Connector configuration registry with lazy instantiation.

Loads connector definitions from typed DeviceConfig objects and materializes them lazily on first access. Supports case-insensitive LFDI lookup.

ConnectorRegistryError

Bases: RuntimeError

Raised when a connector configuration cannot be materialized.

Source code in src/py20305/connectors/registry.py

class ConnectorRegistryError(RuntimeError):
    """Raised when a connector configuration cannot be materialized."""

LazyConnectorProxy

Defers connector construction until the first attribute access.

Source code in src/py20305/connectors/registry.py

class LazyConnectorProxy:
    """Defers connector construction until the first attribute access."""

    def __init__(self, name: str, factory: Callable[[], Any]) -> None:
        # Use object.__setattr__ to avoid triggering __getattr__
        object.__setattr__(self, "_name", name)
        object.__setattr__(self, "_factory", factory)
        object.__setattr__(self, "_instance", None)
        # Lazily-allocated asyncio.Lock that serialises ``aresolve`` calls
        # so two concurrent first-touch awaits don't both spawn the
        # factory (which for ``ConnectorSunSpec`` would mean two TCP
        # connections + two scans against the same Modbus slave). The
        # lock binds to whichever loop first awaited it; rebind if the
        # running loop changes (crash-recovery rebuild + session-scoped
        # test fixtures) -- same pattern as
        # ``ConnectorSunSpec._get_lock``.
        object.__setattr__(self, "_aresolve_lock", None)
        object.__setattr__(self, "_aresolve_lock_loop", None)
        # Cache for permanent construction failures (e.g. Modbus
        # exception codes 1-4 from ``ConnectorSunSpec``'s scan). Without
        # the cache, every poll cycle re-runs the factory and re-scans
        # the broken slave -- on a demo deployment that drove
        # the diagnostics panel counts up by ~6/min per misconfigured
        # device. With the cache, the proxy re-raises the same exception
        # for ``_PERMANENT_FAILURE_TTL_SECONDS`` before letting another
        # construction attempt through, so a device that comes back
        # online (firmware reload, model registration) recovers without
        # a restart.
        object.__setattr__(self, "_permanent_failure", None)
        object.__setattr__(self, "_permanent_failure_at", None)

    def _check_cached_permanent_failure(self) -> None:
        """Re-raise the cached construction failure if it's still fresh.

        Called by both ``resolve`` and ``aresolve`` before running the
        factory. After the TTL expires the cache is cleared so the next
        call attempts construction again -- a device that operators
        repaired sees recovery within ``_PERMANENT_FAILURE_TTL_SECONDS``
        without a restart.
        """
        if self._permanent_failure is None or self._permanent_failure_at is None:
            return
        age = time.monotonic() - self._permanent_failure_at
        if age < _PERMANENT_FAILURE_TTL_SECONDS:
            raise self._permanent_failure
        # TTL expired -- clear and let the factory run again.
        logger.info(
            "LazyConnectorProxy %s: permanent-failure cache expired after %.0fs; "
            "retrying construction",
            self._name,
            age,
        )
        object.__setattr__(self, "_permanent_failure", None)
        object.__setattr__(self, "_permanent_failure_at", None)

    def _record_permanent_failure(self, exc: BaseException) -> None:
        """Cache a permanent construction failure for the TTL window."""
        object.__setattr__(self, "_permanent_failure", exc)
        object.__setattr__(self, "_permanent_failure_at", time.monotonic())

    def resolve(self) -> Any:
        """Return the underlying connector, creating it on demand.

        Synchronous: blocks the calling thread for the duration of the
        factory (which for ``ConnectorSunSpec`` includes a Modbus scan
        with retries). Use ``aresolve`` from the async hot paths so the
        event loop can keep handling other requests during construction.

        Permanent construction failures (factory raised
        ``ConnectorConnectionError(permanent=True)``) are cached for
        ``_PERMANENT_FAILURE_TTL_SECONDS`` and re-raised without
        re-running the factory.
        """
        if self._instance is not None:
            return self._instance
        self._check_cached_permanent_failure()
        try:
            instance = self._factory()
        except Exception as exc:
            if getattr(exc, "permanent", False):
                self._record_permanent_failure(exc)
            raise
        object.__setattr__(self, "_instance", instance)
        return instance

    def _get_aresolve_lock(self) -> asyncio.Lock:
        """Return an ``asyncio.Lock`` bound to the running loop.

        Lazy-allocate on first call; rebind if the loop changes since
        the last allocation. ``asyncio.Lock`` instances bind to the
        loop that first awaited them and raise ``RuntimeError`` if
        reused on another loop -- which happens whenever the registry
        outlives its loop (crash-recovery rebuild; session-scoped
        fixtures in tests).
        """
        loop = asyncio.get_running_loop()
        if self._aresolve_lock is None or self._aresolve_lock_loop is not loop:
            object.__setattr__(self, "_aresolve_lock", asyncio.Lock())
            object.__setattr__(self, "_aresolve_lock_loop", loop)
        return self._aresolve_lock  # type: ignore[no-any-return]

    async def aresolve(self) -> Any:
        """Async resolver: runs ``resolve`` on a worker thread.

        First-touch construction of a real Modbus connector takes seconds
        (TCP handshake + scan + readiness retries). Calling it from an
        ``async def`` route on the client's event loop blocks every
        other coroutine for that whole window. This wrapper offloads the
        work via ``asyncio.to_thread`` so the loop stays responsive.

        Once constructed, returns the cached instance with no thread hop,
        so subsequent calls cost the same as ``resolve``.

        The lock-protected double-check around the to_thread guarantees
        the factory runs at most once per proxy even when N coroutines
        race for first-touch construction. Without it, the per-device
        4 cycle types (metering / capability / settings / status) firing
        close together at startup could each spawn their own
        ``ConnectorSunSpec.__init__`` and open duplicate Modbus
        connections -- the exact fan-out this wrapper exists to make
        safe. Note: ``asyncio.to_thread`` itself
        uses the default ``ThreadPoolExecutor`` (size
        ``min(32, cpu_count + 4)``), so concurrent first-touch awaits
        across *different* proxies can still queue behind the pool;
        wall-clock for first dashboard load with N >> pool_size devices
        is bounded by ``ceil(N / pool_size) * scan_time``.
        """
        if self._instance is not None:
            return self._instance
        # Fast-path: a cached permanent failure within TTL re-raises
        # without taking the lock or hitting a worker thread, so the
        # 4 cycle types polling a known-broken device cost almost
        # nothing per cycle.
        self._check_cached_permanent_failure()
        async with self._get_aresolve_lock():
            # Double-check: another coroutine may have completed the
            # construction while we were waiting on the lock.
            if self._instance is not None:
                return self._instance
            self._check_cached_permanent_failure()
            return await asyncio.to_thread(self.resolve)

    def __getattr__(self, item: str) -> Any:
        return getattr(self.resolve(), item)

    def __repr__(self) -> str:
        state = "initialized" if self._instance is not None else "lazy"
        return f"<LazyConnectorProxy name={self._name!r} state={state}>"

resolve

resolve() -> Any

Return the underlying connector, creating it on demand.

Synchronous: blocks the calling thread for the duration of the factory (which for ConnectorSunSpec includes a Modbus scan with retries). Use aresolve from the async hot paths so the event loop can keep handling other requests during construction.

Permanent construction failures (factory raised ConnectorConnectionError(permanent=True)) are cached for _PERMANENT_FAILURE_TTL_SECONDS and re-raised without re-running the factory.

Source code in src/py20305/connectors/registry.py

def resolve(self) -> Any:
    """Return the underlying connector, creating it on demand.

    Synchronous: blocks the calling thread for the duration of the
    factory (which for ``ConnectorSunSpec`` includes a Modbus scan
    with retries). Use ``aresolve`` from the async hot paths so the
    event loop can keep handling other requests during construction.

    Permanent construction failures (factory raised
    ``ConnectorConnectionError(permanent=True)``) are cached for
    ``_PERMANENT_FAILURE_TTL_SECONDS`` and re-raised without
    re-running the factory.
    """
    if self._instance is not None:
        return self._instance
    self._check_cached_permanent_failure()
    try:
        instance = self._factory()
    except Exception as exc:
        if getattr(exc, "permanent", False):
            self._record_permanent_failure(exc)
        raise
    object.__setattr__(self, "_instance", instance)
    return instance

aresolve async

aresolve() -> Any

Async resolver: runs resolve on a worker thread.

First-touch construction of a real Modbus connector takes seconds (TCP handshake + scan + readiness retries). Calling it from an async def route on the client's event loop blocks every other coroutine for that whole window. This wrapper offloads the work via asyncio.to_thread so the loop stays responsive.

Once constructed, returns the cached instance with no thread hop, so subsequent calls cost the same as resolve.

The lock-protected double-check around the to_thread guarantees the factory runs at most once per proxy even when N coroutines race for first-touch construction. Without it, the per-device 4 cycle types (metering / capability / settings / status) firing close together at startup could each spawn their own ConnectorSunSpec.__init__ and open duplicate Modbus connections -- the exact fan-out this wrapper exists to make safe. Note: asyncio.to_thread itself uses the default ThreadPoolExecutor (size min(32, cpu_count + 4)), so concurrent first-touch awaits across different proxies can still queue behind the pool; wall-clock for first dashboard load with N >> pool_size devices is bounded by ceil(N / pool_size) * scan_time.

Source code in src/py20305/connectors/registry.py

async def aresolve(self) -> Any:
    """Async resolver: runs ``resolve`` on a worker thread.

    First-touch construction of a real Modbus connector takes seconds
    (TCP handshake + scan + readiness retries). Calling it from an
    ``async def`` route on the client's event loop blocks every
    other coroutine for that whole window. This wrapper offloads the
    work via ``asyncio.to_thread`` so the loop stays responsive.

    Once constructed, returns the cached instance with no thread hop,
    so subsequent calls cost the same as ``resolve``.

    The lock-protected double-check around the to_thread guarantees
    the factory runs at most once per proxy even when N coroutines
    race for first-touch construction. Without it, the per-device
    4 cycle types (metering / capability / settings / status) firing
    close together at startup could each spawn their own
    ``ConnectorSunSpec.__init__`` and open duplicate Modbus
    connections -- the exact fan-out this wrapper exists to make
    safe. Note: ``asyncio.to_thread`` itself
    uses the default ``ThreadPoolExecutor`` (size
    ``min(32, cpu_count + 4)``), so concurrent first-touch awaits
    across *different* proxies can still queue behind the pool;
    wall-clock for first dashboard load with N >> pool_size devices
    is bounded by ``ceil(N / pool_size) * scan_time``.
    """
    if self._instance is not None:
        return self._instance
    # Fast-path: a cached permanent failure within TTL re-raises
    # without taking the lock or hitting a worker thread, so the
    # 4 cycle types polling a known-broken device cost almost
    # nothing per cycle.
    self._check_cached_permanent_failure()
    async with self._get_aresolve_lock():
        # Double-check: another coroutine may have completed the
        # construction while we were waiting on the lock.
        if self._instance is not None:
            return self._instance
        self._check_cached_permanent_failure()
        return await asyncio.to_thread(self.resolve)

ConnectorConfigRegistry

Resolves a device LFDI to its connector, constructing it on first use.

Holds typed device configurations and hands out a :class:LazyConnectorProxy per device. Nothing opens a socket or scans a device until something asks for that device's connector, so a configuration naming an offline device costs nothing until it is used.

LFDI lookup is case-insensitive: certificate tooling differs on hex case and a device should resolve either way.

An application with device types of its own supplies a factory_resolver rather than subclassing. That covers the cases class_path cannot: a connector that needs shared runtime state, one reached over a transport this package knows nothing about, or one whose configuration is a model defined downstream.

Source code in src/py20305/connectors/registry.py

class ConnectorConfigRegistry:
    """Resolves a device LFDI to its connector, constructing it on first use.

    Holds typed device configurations and hands out a
    :class:`LazyConnectorProxy` per device. Nothing opens a socket or scans a
    device until something asks for that device's connector, so a
    configuration naming an offline device costs nothing until it is used.

    LFDI lookup is case-insensitive: certificate tooling differs on hex case
    and a device should resolve either way.

    An application with device types of its own supplies a ``factory_resolver``
    rather than subclassing. That covers the cases ``class_path`` cannot: a
    connector that needs shared runtime state, one reached over a transport
    this package knows nothing about, or one whose configuration is a model
    defined downstream.
    """

    def __init__(
        self,
        devices: list[DeviceConfig],
        factory_resolver: Callable[[Any], Callable[[], Any] | None] | None = None,
    ) -> None:
        """Build a registry from typed device configs.

        Args:
            devices: The configured devices. Each is resolved to a connector
                class through :data:`CONNECTOR_TYPE_MAP`, or through its own
                ``class_path`` for
                :class:`~py20305.connectors.config.CustomDeviceConfig`.
            factory_resolver: Consulted for every device before the built-in
                resolution runs. Return a zero-argument callable constructing
                the connector to claim the device, or ``None`` to let this
                package handle it. Called once per device at first resolution,
                not on every lookup, and the callable it returns is invoked
                lazily like any other -- so a resolver may close over
                expensive state without paying for it up front.
        """
        self._device_configs: dict[str, DeviceConfig] = {}
        self._device_proxies: dict[str, LazyConnectorProxy] = {}
        self._factory_resolver = factory_resolver

        for device in devices:
            self._device_configs[device.lfdi] = device

    def get_connector(self, lfdi: str) -> LazyConnectorProxy | None:
        """Return a proxy for the device connector mapped to lfdi.

        LFDI comparison is case-insensitive, and the proxy is cached under the
        configured spelling so it is returned whichever spelling is asked for.
        Caching under the caller's spelling instead would mean a caller using
        a different case never hits the cache -- rebuilding the proxy and the
        connector on every lookup, which for a Modbus device is a fresh
        connection and scan each time, and which defeats the proxy's own
        failure cache and single-flight lock.
        """
        canonical = self._canonical_lfdi(lfdi)
        if canonical is None:
            return None

        cached = self._device_proxies.get(canonical)
        if cached is not None:
            return cached

        proxy = self._build_lazy_proxy(self._device_configs[canonical])
        self._device_proxies[canonical] = proxy
        return proxy

    def _canonical_lfdi(self, lfdi: str) -> str | None:
        """Return the configured spelling of ``lfdi``, or ``None`` if unknown."""
        if lfdi in self._device_configs:
            return lfdi
        wanted = lfdi.lower()
        for stored in self._device_configs:
            if stored.lower() == wanted:
                return stored
        return None

    def iter_device_specs(self) -> list[DeviceConfig]:
        """Return all registered device configurations."""
        return list(self._device_configs.values())

    # ------------------------------------------------------------------
    # Internal
    # ------------------------------------------------------------------

    def _build_lazy_proxy(self, device: DeviceConfig) -> LazyConnectorProxy:
        # Three branches, in precedence order. A downstream ``factory_resolver``
        # gets first refusal, so an application can own device types this
        # package has never heard of. A ``custom`` device carries its own
        # ``class_path`` and ``init_kwargs``, the escape hatch for a connector
        # that needs no shared state. Everything else resolves through the map
        # and unpacks its model into kwargs, minus the fields every device
        # config shares.
        #
        # The proxy wraps all three identically, so a downstream connector gets
        # the same lazy construction, permanent-failure caching and
        # single-flight ``aresolve`` as a bundled one.
        from py20305.connectors.config import CustomDeviceConfig

        if self._factory_resolver is not None:
            resolved = self._factory_resolver(device)
            if resolved is not None:
                return LazyConnectorProxy(name=device.lfdi, factory=resolved)

        if isinstance(device, CustomDeviceConfig):
            class_path = device.class_path
            kwargs = device.init_kwargs
        else:
            try:
                class_path = CONNECTOR_TYPE_MAP[device.type]
            except KeyError as exc:
                known = ", ".join(sorted(CONNECTOR_TYPE_MAP))
                raise ConnectorRegistryError(
                    f"Unknown device type {device.type!r}. Known types: {known}. "
                    "For a connector this package does not ship, either configure "
                    "the device as type 'custom' with a class_path pointing at "
                    "your BaseConnector subclass, or pass a factory_resolver that "
                    "claims this type."
                ) from exc
            kwargs = device.model_dump(exclude={"type", "lfdi", "description", "pin"})

        def factory() -> Any:
            cls = self._import_class(class_path)
            return cls(**kwargs)

        return LazyConnectorProxy(name=device.lfdi, factory=factory)

    def _import_class(self, class_path: str) -> type:
        try:
            module_name, class_name = class_path.rsplit(".", 1)
        except ValueError as exc:
            raise ConnectorRegistryError(f"Invalid class path '{class_path}'.") from exc

        try:
            module = importlib.import_module(module_name)
            return getattr(module, class_name)  # type: ignore[no-any-return]
        except (ImportError, AttributeError) as exc:
            raise ConnectorRegistryError(
                f"Unable to import connector class '{class_path}'."
            ) from exc

get_connector

get_connector(lfdi: str) -> LazyConnectorProxy | None

Return a proxy for the device connector mapped to lfdi.

LFDI comparison is case-insensitive, and the proxy is cached under the configured spelling so it is returned whichever spelling is asked for. Caching under the caller's spelling instead would mean a caller using a different case never hits the cache -- rebuilding the proxy and the connector on every lookup, which for a Modbus device is a fresh connection and scan each time, and which defeats the proxy's own failure cache and single-flight lock.

Source code in src/py20305/connectors/registry.py

def get_connector(self, lfdi: str) -> LazyConnectorProxy | None:
    """Return a proxy for the device connector mapped to lfdi.

    LFDI comparison is case-insensitive, and the proxy is cached under the
    configured spelling so it is returned whichever spelling is asked for.
    Caching under the caller's spelling instead would mean a caller using
    a different case never hits the cache -- rebuilding the proxy and the
    connector on every lookup, which for a Modbus device is a fresh
    connection and scan each time, and which defeats the proxy's own
    failure cache and single-flight lock.
    """
    canonical = self._canonical_lfdi(lfdi)
    if canonical is None:
        return None

    cached = self._device_proxies.get(canonical)
    if cached is not None:
        return cached

    proxy = self._build_lazy_proxy(self._device_configs[canonical])
    self._device_proxies[canonical] = proxy
    return proxy

iter_device_specs

iter_device_specs() -> list[DeviceConfig]

Return all registered device configurations.

Source code in src/py20305/connectors/registry.py

def iter_device_specs(self) -> list[DeviceConfig]:
    """Return all registered device configurations."""
    return list(self._device_configs.values())

py20305.connectors.dispatcher

ConnectorDispatcher: implements ControlDispatcher backed by connectors.

Bridges the event processing system to the connector system. Translates DERControl Pydantic models into connector method calls via the modes translation layer.

ConnectorDispatcher

Implements ControlDispatcher protocol using the connector system.

Resolves device_href to a connector via the registry and an LFDI resolver callback, then translates control models to connector calls.

Source code in src/py20305/connectors/dispatcher.py

class ConnectorDispatcher:
    """Implements ControlDispatcher protocol using the connector system.

    Resolves device_href to a connector via the registry and an LFDI
    resolver callback, then translates control models to connector calls.
    """

    def __init__(
        self,
        registry: ConnectorConfigRegistry,
        lfdi_resolver: Callable[[str], str | None],
        command_observer: CommandObserver | None = None,
        command_gate: CommandGate | None = None,
        telemetry: DeviceTelemetryEmitter | None = None,
    ) -> None:
        """
        Args:
            registry: Connector registry.
            lfdi_resolver: device_href -> LFDI.
            command_observer: Where applied controls are reported. Injected
                rather than imported: the commanded plane is a host-application
                product concern and this module is client code, so the
                dependency has to point the other way. Omitted, controls apply
                exactly as before and nothing is recorded.
            telemetry: Reports each applied control to the monitoring system.
                Optional and disabled by default.
        """
        self._registry = registry
        self._lfdi_resolver = lfdi_resolver
        # `is None`, not `or`: an observer accumulates records, so a
        # collection-backed one is falsy while empty and `or` would silently
        # replace it with the no-op.
        self._gate = AllowAllCommands() if command_gate is None else command_gate
        self._commands = NullCommandObserver() if command_observer is None else command_observer
        self._telemetry = telemetry
        # Devices already reported as having no connector. A server can list a
        # device this process has no connector for, and every control aimed at
        # it reaches the lookup below, so without this the same warning is
        # logged once per control for as long as the device stays listed.
        self._missing_connector_reported: set[str] = set()

    @property
    def telemetry(self) -> DeviceTelemetryEmitter | None:
        """The southbound telemetry emitter, if one was injected.

        The metering cycle reports through the same emitter as the control
        path, so both halves of a device's traffic land on one channel with
        one configuration behind them.
        """
        return self._telemetry

    @property
    def registry(self) -> ConnectorConfigRegistry:
        """The registry this dispatcher resolves through.

        Exposed so a caller wiring a second consumer -- the metering cycle,
        which reads the same devices this writes to -- resolves through the
        same registry. Constructing a second one would build a second
        connector per device and open duplicate connections to it.
        """
        return self._registry

    async def _resolve_connector(self, device_href: str) -> BaseConnector | None:
        """Resolve a device_href to a connector instance.

        Async because first-touch construction of a real Modbus connector
        runs scan + readiness retries that block whichever thread does the
        work; offload to a worker via ``aresolve``.
        """
        from py20305.diagnostics import report

        lfdi = self._lfdi_resolver(device_href)
        if lfdi is None:
            report(
                "warnings",
                f"No LFDI found for device href {device_href}",
                source="dispatcher",
                dedup_key=f"unresolved_href:{device_href}",
                details={"device_href": device_href},
            )
            return None

        return await self._resolve_connector_by_lfdi(lfdi)

    async def _resolve_connector_by_lfdi(self, lfdi: str) -> BaseConnector | None:
        """Resolve a device LFDI directly to a connector instance.

        Used by multi-device dispatch where the processor
        has already expanded a server-side EndDevice href into a list of
        local sub-device LFDIs and we skip the href→LFDI step.
        """
        from py20305.diagnostics import report, resolve

        dedup_key = f"no_connector:{lfdi}"
        proxy = self._registry.get_connector(lfdi)
        if proxy is not None:
            if lfdi in self._missing_connector_reported:
                # The device has a connector now: clear the standing warning,
                # and report again if it ever loses the connector.
                self._missing_connector_reported.discard(lfdi)
                resolve(dedup_key)
            return await proxy.aresolve()  # type: ignore[no-any-return]

        if lfdi in self._missing_connector_reported:
            logger.debug("No connector found for LFDI %s (already reported)", lfdi)
            return None
        self._missing_connector_reported.add(lfdi)
        report(
            "warnings",
            f"No connector found for LFDI {lfdi}; controls for this device are not applied",
            source="dispatcher",
            dedup_key=dedup_key,
            details={"lfdi": lfdi},
        )
        return None

    async def apply_control(
        self,
        device_href: str,
        derc: Dercontrol1,
        curves: list[Dercurve1],
        *,
        origin: str = CommandOrigin.IEEE2030_5,
    ) -> None:
        """Apply a DERControl's settings to a device."""
        connector = await self._resolve_connector(device_href)
        await self._apply_control_to_connector(
            connector,
            derc,
            curves,
            lfdi=self._lfdi_resolver(device_href),
            device_href=device_href,
            origin=origin,
        )

    async def apply_control_by_lfdi(
        self,
        lfdi: str,
        derc: Dercontrol1,
        curves: list[Dercurve1],
        *,
        origin: str = CommandOrigin.IEEE2030_5,
    ) -> None:
        """Apply a DERControl directly to a device identified by LFDI."""
        connector = await self._resolve_connector_by_lfdi(lfdi)
        await self._apply_control_to_connector(
            connector, derc, curves, lfdi=lfdi, device_href=None, origin=origin
        )

    async def _apply_control_to_connector(
        self,
        connector: BaseConnector | None,
        derc: Dercontrol1,
        curves: list[Dercurve1],
        *,
        lfdi: str | None,
        device_href: str | None,
        origin: str = CommandOrigin.IEEE2030_5,
    ) -> None:
        if connector is None:
            return

        label = lfdi or device_href or "<unknown>"
        # An event's own activation: what was written and what caused it are
        # the same event.
        derc_mrid = derc.m_rid.value
        translations = translate_controls(derc.dercontrol_base, curves)
        logger.debug(
            "Applying control to %s via %s (%d mode(s))",
            label,
            connector.connector_name,
            len(translations),
        )

        # Pass ramp_tms through to each mode's params when set
        ramp_tms = derc.dercontrol_base.ramp_tms
        if ramp_tms is not None:
            for _method, p in translations:
                p["ramp_tms"] = ramp_tms

        for method_name, params in translations:
            method, reason = self._control_support(connector, method_name)
            if method is not None:
                logger.debug("  %s.%s(%s)", connector.connector_name, method_name, params)
                await self._apply_one(
                    method,
                    method_name,
                    params,
                    lfdi=lfdi,
                    origin=origin,
                    label=label,
                    applied_mrid=derc_mrid,
                    cause_mrid=derc_mrid,
                )
            elif reason == BY_DESIGN:
                # This connector has no register for the mode and never claimed
                # one. Nothing an operator can do, so it does not belong in a
                # warnings surface -- and a control carrying it alongside modes
                # the connector does implement is normal traffic, not a fault.
                logger.debug(
                    "  %s does not implement %s; skipped",
                    connector.connector_name,
                    method_name,
                )
            else:
                self._report_unimplemented_mode(
                    connector, lfdi=lfdi, device_href=device_href, method_name=method_name
                )

    async def _apply_one(
        self,
        method: Callable[[dict[str, Any]], Any],
        method_name: str,
        params: dict[str, Any],
        *,
        lfdi: str | None,
        origin: str,
        label: str,
        connector: BaseConnector | None = None,
        applied_mrid: bytes | None = None,
        cause_mrid: bytes | None = None,
    ) -> bool:
        """Apply one translated mode and report it to the command observer.

        Returns True when the control reached the connector, False when the gate
        refused it. A caller dispatching a server-issued control ignores the
        result, since a refusal there is reported rather than raised; a caller
        that named one control needs to tell the two apart.

        Stamped before the call, like an acquisition: the write shadow compares
        this against when a readback *started*, so a command has to be marked at
        the moment it was issued rather than when it completed.

        A failure is recorded and re-raised. Swallowing it here would change the
        dispatcher's contract with the event processor, and a rejected command is
        exactly the thing an audit trail should retain.
        """
        # ``control`` drops the ``update_`` prefix so the recorded name matches
        # the management API's operation vocabulary -- one name per control,
        # whichever interface issued it.
        control = method_name.removeprefix("update_")

        # Authority is checked here because this is the one place every apply
        # path passes through -- a control, a default control, a clear and a
        # named operation. Checking at each caller instead would mean the DDERC
        # reapply and comms-loss routes each needing their own gate, which is
        # exactly how a write path ends up ungated.
        if lfdi:
            if not self._gate.may_command(lfdi, origin):
                self._report_not_commanding(lfdi, control, origin)
                # Recorded as a write that did not happen: an event's activation
                # still reports the device as started -- the refusal honors the
                # configuration rather than failing the event -- so without this
                # the trail would show a device as applied with nothing written.
                if self._telemetry is not None:
                    self._telemetry.record_write(
                        label,
                        control,
                        params,
                        connector=(
                            connector
                            if connector is not None
                            else getattr(method, "__self__", None)
                        ),
                        lfdi=lfdi,
                        error=f"{origin} does not hold the command role for this device",
                        error_type="refused",
                        origin=origin,
                        applied_mrid=applied_mrid,
                        cause_mrid=cause_mrid,
                    )
                return False
        else:
            # Deliberately ungated: authority is held per device, and there is no
            # device here to hold it over. Denying instead would drop writes for
            # an href that resolves to a connector but not to an LFDI -- a case
            # that predates the gate and is handled below -- so the choice is to
            # let it through and say so rather than to fail closed by accident.
            logger.debug("Ungated %s on %s: no LFDI to resolve authority for", control, label)
        issued_at = time.time()
        # A bound method carries its connector, so the device's address and
        # protocol are reachable without widening every call site. A caller
        # that wraps the method -- the native clear-control path passes a
        # lambda -- has no ``__self__`` to introspect, so it supplies the
        # connector explicitly and that takes precedence.
        if connector is None:
            connector = getattr(method, "__self__", None)
        try:
            await method(params)
        except Exception as exc:
            if lfdi:
                self._commands.record_command(
                    lfdi, control, params, origin=origin, at=issued_at, error=str(exc)
                )
            # A rejected command is reported too. The northbound side may still
            # believe it succeeded, and that divergence is the point.
            if self._telemetry is not None:
                self._telemetry.record_write(
                    label,
                    control,
                    params,
                    connector=connector,
                    lfdi=lfdi,
                    error=str(exc),
                    error_type=type(exc).__name__,
                    origin=origin,
                    applied_mrid=applied_mrid,
                    cause_mrid=cause_mrid,
                )
            raise
        if self._telemetry is not None:
            self._telemetry.record_write(
                label,
                control,
                params,
                connector=connector,
                lfdi=lfdi,
                origin=origin,
                applied_mrid=applied_mrid,
                cause_mrid=cause_mrid,
            )
        if lfdi:
            self._commands.record_command(lfdi, control, params, origin=origin, at=issued_at)
        else:
            # No LFDI means no key to file the record under. Rare (an href that
            # resolves to a connector but not to an LFDI) and worth seeing.
            logger.debug("Applied %s to %s with no LFDI; not recorded", control, label)
        return True

    async def apply_operation(
        self,
        lfdi: str,
        control: str,
        params: dict[str, Any],
        *,
        origin: str,
    ) -> None:
        """Apply one named control to one device by LFDI, and record it.

        The write funnel for callers that already know which control they want,
        rather than a DERControl to be translated -- a Modbus master writing a
        register, for instance. Everything an event-driven write gets applies:
        the command is recorded with its origin, a failure is recorded as
        rejected and re-raised.

        Raises:
            ConnectorError: no connector is registered for ``lfdi``, or the
                connector does not implement this control.
            CommandNotPermittedError: a gate refused this origin authority over
                this device. Raised rather than reported because this caller
                named the control and has somewhere to put the answer -- a
                protocol server owes its own client the distinction between a
                write that failed and one that was not permitted.
        """
        connector = await self._resolve_connector_by_lfdi(lfdi)
        if connector is None:
            raise ConnectorError(f"no connector for LFDI {lfdi}")
        method_name = f"update_{control}"
        method, _reason = self._control_support(connector, method_name)
        if method is None:
            raise ConnectorError(f"connector for {lfdi} does not implement {method_name}")
        applied = await self._apply_one(
            method, method_name, params, lfdi=lfdi, origin=origin, label=lfdi
        )
        if not applied:
            raise CommandNotPermittedError(
                f"{origin} may not command {lfdi}: another interface holds the "
                f"command role, so {control!r} was not applied"
            )

    async def apply_default_control(
        self,
        device_href: str,
        dderc: DefaultDercontrol,
        curves: list[Dercurve1],
        *,
        origin: str = CommandOrigin.DDERC_REAPPLY,
        cause_mrid: bytes | None = None,
    ) -> None:
        """Apply default DER control (DDERC fallback) to a device.

        ``cause_mrid`` names the event whose end brought the default back, for
        the audit trail. ``None`` when no single event caused it.
        """
        connector = await self._resolve_connector(device_href)
        await self._apply_default_control_to_connector(
            connector,
            dderc,
            curves,
            lfdi=self._lfdi_resolver(device_href),
            device_href=device_href,
            origin=origin,
            cause_mrid=cause_mrid,
        )

    async def apply_default_control_by_lfdi(
        self,
        lfdi: str,
        dderc: DefaultDercontrol,
        curves: list[Dercurve1],
        *,
        origin: str = CommandOrigin.DDERC_REAPPLY,
        cause_mrid: bytes | None = None,
    ) -> None:
        """Apply default DER control directly to a device identified by LFDI."""
        connector = await self._resolve_connector_by_lfdi(lfdi)
        await self._apply_default_control_to_connector(
            connector,
            dderc,
            curves,
            lfdi=lfdi,
            device_href=None,
            origin=origin,
            cause_mrid=cause_mrid,
        )

    async def _apply_default_control_to_connector(
        self,
        connector: BaseConnector | None,
        dderc: DefaultDercontrol,
        curves: list[Dercurve1],
        *,
        lfdi: str | None,
        device_href: str | None,
        origin: str = CommandOrigin.DDERC_REAPPLY,
        cause_mrid: bytes | None = None,
    ) -> None:
        if connector is None:
            return

        label = lfdi or device_href or "<unknown>"
        dderc_mrid = dderc.m_rid.value
        translations = translate_default_controls(dderc, curves)
        logger.debug("Applying DDERC fallback to %s (%d mode(s))", label, len(translations))
        for method_name, params in translations:
            method, reason = self._control_support(connector, method_name)
            if method is not None:
                await self._apply_one(
                    method,
                    method_name,
                    params,
                    lfdi=lfdi,
                    origin=origin,
                    label=label,
                    applied_mrid=dderc_mrid,
                    cause_mrid=cause_mrid,
                )
            elif reason == BY_DESIGN:
                # This connector has no register for the mode and never claimed
                # one. Nothing an operator can do, so it does not belong in a
                # warnings surface -- and a control carrying it alongside modes
                # the connector does implement is normal traffic, not a fault.
                logger.debug(
                    "  %s does not implement %s; skipped",
                    connector.connector_name,
                    method_name,
                )
            else:
                self._report_unimplemented_mode(
                    connector, lfdi=lfdi, device_href=device_href, method_name=method_name
                )

    async def relay_schedule_notification(
        self,
        lfdis: list[str],
        notification: ScheduleNotification,
    ) -> None:
        """Relay a schedule notification to each connector owning an affected LFDI.

        Resolves each LFDI to its connector (the same ``registry.get_connector``
        path dispatch uses), de-duplicates so a connector that owns several affected
        LFDIs is notified once, and isolates per-connector errors -- a
        throwing/slow connector here must never affect event processing or the
        other connectors.

        De-dup is keyed by ``connector.relay_group`` when present. A connector
        fronting several devices over one connection sets it to that
        connection's name, so N per-LFDI instances collapse to a single
        notification -- which is correct, because the notification already
        carries every affected LFDI. Connectors that don't set it fall back to
        instance identity.

        The de-duped connectors are notified **concurrently** so one slow
        connector can't delay delivery to the others; each call is
        error-isolated.
        """
        from py20305.diagnostics import report

        seen: set[object] = set()
        targets: list[tuple[str, BaseConnector]] = []
        for lfdi in lfdis:
            try:
                connector = await self._resolve_connector_by_lfdi(lfdi)
            except asyncio.CancelledError:
                raise
            except Exception as exc:
                # Resolution (e.g. LazyConnectorProxy.aresolve permanent errors)
                # must not abort the whole relay -- isolate and keep going.
                report(
                    "warnings",
                    (
                        f"Resolving connector for {lfdi} failed during schedule relay "
                        f"({notification.stream}/{notification.transition}): {exc}"
                    ),
                    source="dispatcher",
                    dedup_key=f"schedule_notify_resolve:{lfdi}:{notification.stream}",
                    details={"lfdi": lfdi, "stream": notification.stream, "error": str(exc)},
                )
                continue
            if connector is None:
                continue
            group = getattr(connector, "relay_group", None) or id(connector)
            if group in seen:
                continue
            seen.add(group)
            targets.append((lfdi, connector))

        if targets:
            await asyncio.gather(
                *(self._notify_one(lfdi, connector, notification) for lfdi, connector in targets)
            )

    async def _notify_one(
        self,
        lfdi: str,
        connector: BaseConnector,
        notification: ScheduleNotification,
    ) -> None:
        """Relay to a single connector, isolating any error.

        An informational relay must never propagate into event processing or
        affect the other connectors (gather sees no exception).
        """
        from py20305.diagnostics import report

        logger.debug(
            "relay %s/%s mrid=%s -> %s (lfdi=%s, affects=%d)",
            notification.stream,
            notification.transition,
            notification.mrid,
            connector.connector_name,
            lfdi,
            len(notification.affected_lfdis),
        )
        try:
            await connector.on_schedule_notification(notification)
        except asyncio.CancelledError:
            # Shutdown cancels in-flight relay tasks -- a cancellation is not a
            # connector failure; stay quiet and propagate.
            raise
        except Exception as exc:
            report(
                "warnings",
                (
                    f"Connector {connector.connector_name} raised in "
                    f"on_schedule_notification ({notification.stream}/"
                    f"{notification.transition}): {exc}"
                ),
                source="dispatcher",
                dedup_key=f"schedule_notify:{lfdi}:{notification.stream}:{notification.mrid}",
                details={
                    "lfdi": lfdi,
                    "stream": notification.stream,
                    "mrid": notification.mrid,
                    "transition": notification.transition,
                    "error": str(exc),
                },
            )

    @staticmethod
    def _report_not_commanding(lfdi: str, control: str, origin: str) -> None:
        """Surface a write refused because its origin does not command the device.

        Reported rather than raised: a 2030.5 server posting events to a device
        that SunSpec commands is a configuration statement being honored, not a
        fault, and raising would turn every such event into an error the event
        engine has to interpret. Silence is the wrong answer too -- an operator
        seeing a posted control never arrive needs this row to know why.
        """
        from py20305.diagnostics import report

        report(
            "warnings",
            f"{origin} may not command {lfdi[:8]}: another interface holds "
            f"the command role. '{control}' was not applied.",
            source="dispatcher",
            dedup_key=f"not-commanding-{origin}-{lfdi}",
            details={"device": lfdi, "origin": str(origin), "control": control},
        )

    @staticmethod
    def _control_support(
        connector: BaseConnector, method_name: str
    ) -> tuple[Callable[[dict[str, Any]], Any] | None, str | None]:
        """``(bound method, reason)``. ``reason`` is None when implemented.

        ``BaseConnector`` declares 40 ``update_*`` modes as concrete methods that
        return ``None``, so ``getattr`` finds one whether or not the connector
        implements it. Reading that as support makes a connector look like it
        accepted a command it never carried out: the dispatch succeeds, the
        commanded plane records it, and a Modbus master is acknowledged for a
        write that reached no device.

        The two ways a mode can be unsupported need telling apart, because only
        one of them is anybody's mistake:

        ``BY_DESIGN``
            The connector inherits the base no-op. ``ConnectorSunSpec`` does this
            for 29 of the 40 modes and says so in its docstring -- the base
            defaults exist precisely to cover modes SunSpec has no register for.
            Expected, and not something an operator can act on.

        ``OFFER_MISSING``
            No implementation resolves at all. For a plugin-backed connector,
            whose modes arrive through ``__getattr__`` from a live offer, that
            means the offer lacks a mode it should carry -- typically a plugin
            built against an older SDK. Actionable, and what the unimplemented
            diagnostic was written for.
        """
        own = getattr(type(connector), method_name, None)
        if own is not None and own is getattr(BaseConnector, method_name, None):
            return None, BY_DESIGN
        bound: Callable[[dict[str, Any]], Any] | None = getattr(connector, method_name, None)
        if bound is None:
            return None, OFFER_MISSING
        return bound, None

    @staticmethod
    def _report_unimplemented_mode(
        connector: BaseConnector,
        *,
        lfdi: str | None,
        device_href: str | None,
        method_name: str,
    ) -> None:
        """Surface a translated mode the connector can't accept.

        When a server-issued DERControl translates to ``method_name`` but the
        resolved connector doesn't expose that method, the control has nowhere
        to go. Dropping it silently is the wrong answer: the operator sees the
        post land at the client and never reach the device, with nothing
        saying why. The usual cause is a connector built against an older
        revision of this package that predates the mode.

        Routed through ``diagnostics.report`` so it surfaces alongside
        ``no_connector`` and ``unresolved_href``. The dedup key is scoped per
        (LFDI-or-href, method), so a device polled repeatedly collapses to one
        entry with ``count`` ticking up rather than flooding the log.
        """
        from py20305.diagnostics import report

        identifier = lfdi or device_href or "<unknown>"
        report(
            "warnings",
            (
                f"Connector {connector.connector_name} for {identifier} "
                f"does not implement {method_name}; control dropped. "
                "A connector supplied outside this package may predate "
                "this mode -- rebuild it against the current version."
            ),
            source="dispatcher",
            dedup_key=f"unimplemented_mode:{identifier}:{method_name}",
            details={
                "lfdi": lfdi,
                "device_href": device_href,
                "method": method_name,
                "connector": connector.connector_name,
            },
        )

    async def clear_control(
        self,
        device_href: str,
        *,
        cause_mrid: bytes | None = None,
    ) -> None:
        """Clear active control from a device by sending disable params.

        Prefers the connector's own ``clear_control`` when it has one. A
        connector talking to a remote device should implement it as a single
        round-trip rather than paying for the per-mode fan-out below; the
        bundled connectors talk to the device directly, so they don't
        override it and take the fan-out.
        """
        logger.debug("Clearing all controls on %s", device_href)
        connector = await self._resolve_connector(device_href)
        await self._clear_control_on_connector(
            connector,
            lfdi=self._lfdi_resolver(device_href),
            label=device_href,
            cause_mrid=cause_mrid,
        )

    async def clear_control_by_lfdi(
        self,
        lfdi: str,
        *,
        cause_mrid: bytes | None = None,
    ) -> None:
        """Clear active control from a device identified by LFDI.

        The by-LFDI counterpart of ``clear_control`` (the comms-loss
        safe-default when dispatching by LFDI). Skips the href->LFDI
        resolution step.
        """
        logger.debug("Clearing all controls on LFDI %s", lfdi)
        connector = await self._resolve_connector_by_lfdi(lfdi)
        await self._clear_control_on_connector(
            connector, lfdi=lfdi, label=lfdi, cause_mrid=cause_mrid
        )

    async def disable_modes(
        self,
        device_href: str,
        modes: frozenset[str],
        *,
        cause_mrid: bytes | None = None,
    ) -> None:
        """Disable the named control modes on a device.

        ``modes`` are DERControlBase field names (``op_mod_fixed_w``) or the
        CSIP-AUS limit element names (``opModExpLimW``), as the event processor
        tracks them. Called at event end for the modes the ended event set and
        the DefaultDERControl does not carry.
        """
        connector = await self._resolve_connector(device_href)
        await self._disable_modes_on_connector(
            connector,
            modes,
            lfdi=self._lfdi_resolver(device_href),
            label=device_href,
            cause_mrid=cause_mrid,
        )

    async def disable_modes_by_lfdi(
        self,
        lfdi: str,
        modes: frozenset[str],
        *,
        cause_mrid: bytes | None = None,
    ) -> None:
        """The by-LFDI counterpart of ``disable_modes``."""
        connector = await self._resolve_connector_by_lfdi(lfdi)
        await self._disable_modes_on_connector(
            connector, modes, lfdi=lfdi, label=lfdi, cause_mrid=cause_mrid
        )

    async def _disable_modes_on_connector(
        self,
        connector: BaseConnector | None,
        modes: frozenset[str],
        *,
        lfdi: str | None,
        label: str,
        cause_mrid: bytes | None,
    ) -> None:
        if connector is None:
            return
        # One disable per method: two fields can share a method (the two fixed
        # power factor fields both reach ``update_const_pf``).
        calls: dict[str, dict[str, Any]] = {}
        for mode in sorted(modes):
            entry = FIELD_DISABLE_CALLS.get(mode)
            if entry is not None:
                calls.setdefault(entry[0], entry[1])
        for method_name, params in calls.items():
            method, _reason = self._control_support(connector, method_name)
            if method is None:
                continue
            logger.debug("  %s: %s disabled at event end", label, method_name)
            await self._apply_one(
                method,
                method_name,
                params,
                lfdi=lfdi,
                origin=CommandOrigin.EVENT_END,
                label=label,
                cause_mrid=cause_mrid,
            )

    async def _clear_control_on_connector(
        self,
        connector: BaseConnector | None,
        *,
        lfdi: str | None = None,
        label: str = "",
        cause_mrid: bytes | None = None,
    ) -> None:
        """Disable every active control on a device.

        Both branches route through ``_apply_one`` so a clear is recorded like
        any other write. It reaches the device the same way, and it is the
        comms-loss safe default -- the one control action most worth being able
        to account for after the fact -- so leaving it out of the audit trail
        and the monitoring stream would be the wrong asymmetry.
        """
        if connector is None:
            return

        native_clear = getattr(connector, "clear_control", None)
        if callable(native_clear):
            # One round-trip that clears everything, so it is one record
            # rather than the fan-out's per-mode ones.
            await self._apply_one(
                lambda _params: native_clear(),
                "clear_control",
                {},
                lfdi=lfdi,
                origin=CommandOrigin.COMMS_LOSS,
                label=label,
                # The lambda has no ``__self__``; without this the clear would
                # report as generic with no destination -- losing exactly the
                # attribution this path most needs.
                connector=connector,
                cause_mrid=cause_mrid,
            )
            return

        for method_name, params in DISABLE_CALLS:
            # Same distinction the translation loops draw, and it matters more
            # here: this fan-out touches every mode, and a clear is recorded like
            # any other write. Taking an inherited no-op for an implementation
            # would file twenty commands that reached no register, for a device
            # that implements a handful of modes.
            method, _reason = self._control_support(connector, method_name)
            if method is not None:
                await self._apply_one(
                    method,
                    method_name,
                    params,
                    lfdi=lfdi,
                    origin=CommandOrigin.COMMS_LOSS,
                    label=label,
                    cause_mrid=cause_mrid,
                )

telemetry property

telemetry: DeviceTelemetryEmitter | None

The southbound telemetry emitter, if one was injected.

The metering cycle reports through the same emitter as the control path, so both halves of a device's traffic land on one channel with one configuration behind them.

registry property

registry: ConnectorConfigRegistry

The registry this dispatcher resolves through.

Exposed so a caller wiring a second consumer -- the metering cycle, which reads the same devices this writes to -- resolves through the same registry. Constructing a second one would build a second connector per device and open duplicate connections to it.

apply_control async

apply_control(device_href: str, derc: Dercontrol1, curves: list[Dercurve1], *, origin: str = CommandOrigin.IEEE2030_5) -> None

Apply a DERControl's settings to a device.

Source code in src/py20305/connectors/dispatcher.py

async def apply_control(
    self,
    device_href: str,
    derc: Dercontrol1,
    curves: list[Dercurve1],
    *,
    origin: str = CommandOrigin.IEEE2030_5,
) -> None:
    """Apply a DERControl's settings to a device."""
    connector = await self._resolve_connector(device_href)
    await self._apply_control_to_connector(
        connector,
        derc,
        curves,
        lfdi=self._lfdi_resolver(device_href),
        device_href=device_href,
        origin=origin,
    )

apply_control_by_lfdi async

apply_control_by_lfdi(lfdi: str, derc: Dercontrol1, curves: list[Dercurve1], *, origin: str = CommandOrigin.IEEE2030_5) -> None

Apply a DERControl directly to a device identified by LFDI.

Source code in src/py20305/connectors/dispatcher.py

async def apply_control_by_lfdi(
    self,
    lfdi: str,
    derc: Dercontrol1,
    curves: list[Dercurve1],
    *,
    origin: str = CommandOrigin.IEEE2030_5,
) -> None:
    """Apply a DERControl directly to a device identified by LFDI."""
    connector = await self._resolve_connector_by_lfdi(lfdi)
    await self._apply_control_to_connector(
        connector, derc, curves, lfdi=lfdi, device_href=None, origin=origin
    )

apply_operation async

apply_operation(lfdi: str, control: str, params: dict[str, Any], *, origin: str) -> None

Apply one named control to one device by LFDI, and record it.

The write funnel for callers that already know which control they want, rather than a DERControl to be translated -- a Modbus master writing a register, for instance. Everything an event-driven write gets applies: the command is recorded with its origin, a failure is recorded as rejected and re-raised.

Raises:

Type	Description
ConnectorError	no connector is registered for lfdi, or the connector does not implement this control.
CommandNotPermittedError	a gate refused this origin authority over this device. Raised rather than reported because this caller named the control and has somewhere to put the answer -- a protocol server owes its own client the distinction between a write that failed and one that was not permitted.

Source code in src/py20305/connectors/dispatcher.py

async def apply_operation(
    self,
    lfdi: str,
    control: str,
    params: dict[str, Any],
    *,
    origin: str,
) -> None:
    """Apply one named control to one device by LFDI, and record it.

    The write funnel for callers that already know which control they want,
    rather than a DERControl to be translated -- a Modbus master writing a
    register, for instance. Everything an event-driven write gets applies:
    the command is recorded with its origin, a failure is recorded as
    rejected and re-raised.

    Raises:
        ConnectorError: no connector is registered for ``lfdi``, or the
            connector does not implement this control.
        CommandNotPermittedError: a gate refused this origin authority over
            this device. Raised rather than reported because this caller
            named the control and has somewhere to put the answer -- a
            protocol server owes its own client the distinction between a
            write that failed and one that was not permitted.
    """
    connector = await self._resolve_connector_by_lfdi(lfdi)
    if connector is None:
        raise ConnectorError(f"no connector for LFDI {lfdi}")
    method_name = f"update_{control}"
    method, _reason = self._control_support(connector, method_name)
    if method is None:
        raise ConnectorError(f"connector for {lfdi} does not implement {method_name}")
    applied = await self._apply_one(
        method, method_name, params, lfdi=lfdi, origin=origin, label=lfdi
    )
    if not applied:
        raise CommandNotPermittedError(
            f"{origin} may not command {lfdi}: another interface holds the "
            f"command role, so {control!r} was not applied"
        )

apply_default_control async

apply_default_control(device_href: str, dderc: DefaultDercontrol, curves: list[Dercurve1], *, origin: str = CommandOrigin.DDERC_REAPPLY, cause_mrid: bytes | None = None) -> None

Apply default DER control (DDERC fallback) to a device.

cause_mrid names the event whose end brought the default back, for the audit trail. None when no single event caused it.

Source code in src/py20305/connectors/dispatcher.py

async def apply_default_control(
    self,
    device_href: str,
    dderc: DefaultDercontrol,
    curves: list[Dercurve1],
    *,
    origin: str = CommandOrigin.DDERC_REAPPLY,
    cause_mrid: bytes | None = None,
) -> None:
    """Apply default DER control (DDERC fallback) to a device.

    ``cause_mrid`` names the event whose end brought the default back, for
    the audit trail. ``None`` when no single event caused it.
    """
    connector = await self._resolve_connector(device_href)
    await self._apply_default_control_to_connector(
        connector,
        dderc,
        curves,
        lfdi=self._lfdi_resolver(device_href),
        device_href=device_href,
        origin=origin,
        cause_mrid=cause_mrid,
    )

apply_default_control_by_lfdi async

apply_default_control_by_lfdi(lfdi: str, dderc: DefaultDercontrol, curves: list[Dercurve1], *, origin: str = CommandOrigin.DDERC_REAPPLY, cause_mrid: bytes | None = None) -> None

Apply default DER control directly to a device identified by LFDI.

Source code in src/py20305/connectors/dispatcher.py

async def apply_default_control_by_lfdi(
    self,
    lfdi: str,
    dderc: DefaultDercontrol,
    curves: list[Dercurve1],
    *,
    origin: str = CommandOrigin.DDERC_REAPPLY,
    cause_mrid: bytes | None = None,
) -> None:
    """Apply default DER control directly to a device identified by LFDI."""
    connector = await self._resolve_connector_by_lfdi(lfdi)
    await self._apply_default_control_to_connector(
        connector,
        dderc,
        curves,
        lfdi=lfdi,
        device_href=None,
        origin=origin,
        cause_mrid=cause_mrid,
    )

relay_schedule_notification async

relay_schedule_notification(lfdis: list[str], notification: ScheduleNotification) -> None

Relay a schedule notification to each connector owning an affected LFDI.

Resolves each LFDI to its connector (the same registry.get_connector path dispatch uses), de-duplicates so a connector that owns several affected LFDIs is notified once, and isolates per-connector errors -- a throwing/slow connector here must never affect event processing or the other connectors.

De-dup is keyed by connector.relay_group when present. A connector fronting several devices over one connection sets it to that connection's name, so N per-LFDI instances collapse to a single notification -- which is correct, because the notification already carries every affected LFDI. Connectors that don't set it fall back to instance identity.

The de-duped connectors are notified concurrently so one slow connector can't delay delivery to the others; each call is error-isolated.

Source code in src/py20305/connectors/dispatcher.py

async def relay_schedule_notification(
    self,
    lfdis: list[str],
    notification: ScheduleNotification,
) -> None:
    """Relay a schedule notification to each connector owning an affected LFDI.

    Resolves each LFDI to its connector (the same ``registry.get_connector``
    path dispatch uses), de-duplicates so a connector that owns several affected
    LFDIs is notified once, and isolates per-connector errors -- a
    throwing/slow connector here must never affect event processing or the
    other connectors.

    De-dup is keyed by ``connector.relay_group`` when present. A connector
    fronting several devices over one connection sets it to that
    connection's name, so N per-LFDI instances collapse to a single
    notification -- which is correct, because the notification already
    carries every affected LFDI. Connectors that don't set it fall back to
    instance identity.

    The de-duped connectors are notified **concurrently** so one slow
    connector can't delay delivery to the others; each call is
    error-isolated.
    """
    from py20305.diagnostics import report

    seen: set[object] = set()
    targets: list[tuple[str, BaseConnector]] = []
    for lfdi in lfdis:
        try:
            connector = await self._resolve_connector_by_lfdi(lfdi)
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            # Resolution (e.g. LazyConnectorProxy.aresolve permanent errors)
            # must not abort the whole relay -- isolate and keep going.
            report(
                "warnings",
                (
                    f"Resolving connector for {lfdi} failed during schedule relay "
                    f"({notification.stream}/{notification.transition}): {exc}"
                ),
                source="dispatcher",
                dedup_key=f"schedule_notify_resolve:{lfdi}:{notification.stream}",
                details={"lfdi": lfdi, "stream": notification.stream, "error": str(exc)},
            )
            continue
        if connector is None:
            continue
        group = getattr(connector, "relay_group", None) or id(connector)
        if group in seen:
            continue
        seen.add(group)
        targets.append((lfdi, connector))

    if targets:
        await asyncio.gather(
            *(self._notify_one(lfdi, connector, notification) for lfdi, connector in targets)
        )

clear_control async

clear_control(device_href: str, *, cause_mrid: bytes | None = None) -> None

Clear active control from a device by sending disable params.

Prefers the connector's own clear_control when it has one. A connector talking to a remote device should implement it as a single round-trip rather than paying for the per-mode fan-out below; the bundled connectors talk to the device directly, so they don't override it and take the fan-out.

Source code in src/py20305/connectors/dispatcher.py

async def clear_control(
    self,
    device_href: str,
    *,
    cause_mrid: bytes | None = None,
) -> None:
    """Clear active control from a device by sending disable params.

    Prefers the connector's own ``clear_control`` when it has one. A
    connector talking to a remote device should implement it as a single
    round-trip rather than paying for the per-mode fan-out below; the
    bundled connectors talk to the device directly, so they don't
    override it and take the fan-out.
    """
    logger.debug("Clearing all controls on %s", device_href)
    connector = await self._resolve_connector(device_href)
    await self._clear_control_on_connector(
        connector,
        lfdi=self._lfdi_resolver(device_href),
        label=device_href,
        cause_mrid=cause_mrid,
    )

clear_control_by_lfdi async

clear_control_by_lfdi(lfdi: str, *, cause_mrid: bytes | None = None) -> None

Clear active control from a device identified by LFDI.

The by-LFDI counterpart of clear_control (the comms-loss safe-default when dispatching by LFDI). Skips the href->LFDI resolution step.

Source code in src/py20305/connectors/dispatcher.py

async def clear_control_by_lfdi(
    self,
    lfdi: str,
    *,
    cause_mrid: bytes | None = None,
) -> None:
    """Clear active control from a device identified by LFDI.

    The by-LFDI counterpart of ``clear_control`` (the comms-loss
    safe-default when dispatching by LFDI). Skips the href->LFDI
    resolution step.
    """
    logger.debug("Clearing all controls on LFDI %s", lfdi)
    connector = await self._resolve_connector_by_lfdi(lfdi)
    await self._clear_control_on_connector(
        connector, lfdi=lfdi, label=lfdi, cause_mrid=cause_mrid
    )

disable_modes async

disable_modes(device_href: str, modes: frozenset[str], *, cause_mrid: bytes | None = None) -> None

Disable the named control modes on a device.

modes are DERControlBase field names (op_mod_fixed_w) or the CSIP-AUS limit element names (opModExpLimW), as the event processor tracks them. Called at event end for the modes the ended event set and the DefaultDERControl does not carry.

Source code in src/py20305/connectors/dispatcher.py

async def disable_modes(
    self,
    device_href: str,
    modes: frozenset[str],
    *,
    cause_mrid: bytes | None = None,
) -> None:
    """Disable the named control modes on a device.

    ``modes`` are DERControlBase field names (``op_mod_fixed_w``) or the
    CSIP-AUS limit element names (``opModExpLimW``), as the event processor
    tracks them. Called at event end for the modes the ended event set and
    the DefaultDERControl does not carry.
    """
    connector = await self._resolve_connector(device_href)
    await self._disable_modes_on_connector(
        connector,
        modes,
        lfdi=self._lfdi_resolver(device_href),
        label=device_href,
        cause_mrid=cause_mrid,
    )

disable_modes_by_lfdi async

disable_modes_by_lfdi(lfdi: str, modes: frozenset[str], *, cause_mrid: bytes | None = None) -> None

The by-LFDI counterpart of disable_modes.

Source code in src/py20305/connectors/dispatcher.py

async def disable_modes_by_lfdi(
    self,
    lfdi: str,
    modes: frozenset[str],
    *,
    cause_mrid: bytes | None = None,
) -> None:
    """The by-LFDI counterpart of ``disable_modes``."""
    connector = await self._resolve_connector_by_lfdi(lfdi)
    await self._disable_modes_on_connector(
        connector, modes, lfdi=lfdi, label=lfdi, cause_mrid=cause_mrid
    )

py20305.connectors.errors

Errors a connector raises when talking to a device.

Transport-neutral: a connector speaking Modbus, MQTT, SQL or anything else raises these same types, so the code above it -- the dispatcher, the event engine, the API -- reasons about failure without knowing the protocol underneath. :mod:py20305.connectors.control_errors covers the other half: a control the device declined rather than a transport that broke.

ConnectorError

Bases: Exception

Base exception for connector operations.

Source code in src/py20305/connectors/errors.py

class ConnectorError(Exception):
    """Base exception for connector operations."""

ConnectorConnectionError

Bases: ConnectorError

Raised when the connector cannot reach the target device.

The optional permanent flag signals that the failure won't change on retry without operator intervention -- e.g. Modbus protocol exception codes 1-4 (illegal function/address/value, server failure), which typically mean the configured device is missing models the connector was told to scan for.

Source code in src/py20305/connectors/errors.py

class ConnectorConnectionError(ConnectorError):
    """Raised when the connector cannot reach the target device.

    The optional ``permanent`` flag signals that the failure won't change
    on retry without operator intervention -- e.g. Modbus protocol
    exception codes 1-4 (illegal function/address/value, server failure),
    which typically mean the configured device is missing models the
    connector was told to scan for.
    """

    def __init__(self, *args: object, permanent: bool = False) -> None:
        super().__init__(*args)
        self.permanent = permanent

ConnectorTimeoutError

Bases: ConnectorError

Raised when a connector operation exceeds its deadline.

Source code in src/py20305/connectors/errors.py

class ConnectorTimeoutError(ConnectorError):
    """Raised when a connector operation exceeds its deadline."""

ConnectorWriteError

Bases: ConnectorError

Raised when a control write fails or is rejected by the device.

Source code in src/py20305/connectors/errors.py

class ConnectorWriteError(ConnectorError):
    """Raised when a control write fails or is rejected by the device."""

ConnectorValueError

Bases: ConnectorError

Raised when a control parameter is outside the range the profile allows.

Distinct from :class:ConnectorWriteError: nothing was attempted. The parameter itself is invalid, so no device could have accepted it and a retry with the same value cannot succeed. The register is left untouched, including any mode-enable point -- a lever must not be raised for a write that will not happen.

Reported to the head-end as IEEE 2030.5 Table 31 status 253 (Invalid), because the fault is in the event's data rather than in what the device can do. Adapters translate accordingly at their own boundary.

Source code in src/py20305/connectors/errors.py

class ConnectorValueError(ConnectorError):
    """Raised when a control parameter is outside the range the profile allows.

    Distinct from :class:`ConnectorWriteError`: nothing was attempted. The
    parameter itself is invalid, so no device could have accepted it and a retry
    with the same value cannot succeed. The register is left untouched, including
    any mode-enable point -- a lever must not be raised for a write that will not
    happen.

    Reported to the head-end as IEEE 2030.5 Table 31 status 253 (Invalid),
    because the fault is in the event's data rather than in what the device can
    do. Adapters translate accordingly at their own boundary.
    """

py20305.connectors.control_errors

Errors a connector raises when it will not perform a control.

Distinct from :mod:py20305.connectors.errors, which covers a transport that broke. These say the transport was fine and the control still did not happen -- the device is offline, the mode is unsupported, the value is out of range, or the site declined on the customer's behalf.

The distinction is load-bearing because IEEE 2030.5 makes the head-end care about it. Each type below carries the retry behavior the caller should apply and the Table 31 response status the head-end should see; see :func:py20305.events.response.response_code_for_dispatch_error for the mapping.

ControlError

Bases: Exception

Base for a control a connector declined to perform.

Source code in src/py20305/connectors/control_errors.py

class ControlError(Exception):
    """Base for a control a connector declined to perform."""

DeviceOfflineTransientError

Bases: ControlError

The device is unreachable, but may return on its own.

Caller logs and retries on the next polling cycle.

Source code in src/py20305/connectors/control_errors.py

class DeviceOfflineTransientError(ControlError):
    """The device is unreachable, but may return on its own.

    Caller logs and retries on the next polling cycle.
    """

DeviceOfflinePermanentError

Bases: ControlError

The device is unreachable and will stay that way without intervention.

Caller suppresses further attempts for this device rather than retrying every cycle.

Source code in src/py20305/connectors/control_errors.py

class DeviceOfflinePermanentError(ControlError):
    """The device is unreachable and will stay that way without intervention.

    Caller suppresses further attempts for this device rather than retrying
    every cycle.
    """

DeviceNotConfiguredError

Bases: ControlError

The connector has no configuration for the addressed device.

Caller re-sends its device configuration once and retries.

Source code in src/py20305/connectors/control_errors.py

class DeviceNotConfiguredError(ControlError):
    """The connector has no configuration for the addressed device.

    Caller re-sends its device configuration once and retries.
    """

ModeNotSupportedError

Bases: ControlError

The device cannot perform the requested control mode.

Permanent for this device and mode. Callers that pre-filter by a device's advertised modes should never see this; if one does, the advertised set and the device disagree.

Source code in src/py20305/connectors/control_errors.py

class ModeNotSupportedError(ControlError):
    """The device cannot perform the requested control mode.

    Permanent for this device and mode. Callers that pre-filter by a device's
    advertised modes should never see this; if one does, the advertised set and
    the device disagree.
    """

OptOutError

Bases: ControlError

The control was declined on the customer's behalf.

Reports a decision, not a failure, and is surfaced to the head-end as IEEE 2030.5 Table 31 status 4 rather than a capability-limit code.

Source code in src/py20305/connectors/control_errors.py

class OptOutError(ControlError):
    """The control was declined on the customer's behalf.

    Reports a decision, not a failure, and is surfaced to the head-end as
    IEEE 2030.5 Table 31 status 4 rather than a capability-limit code.
    """

InvalidControlValueError

Bases: ControlError

A control parameter is outside the range the IEEE 2030.5 profile allows.

Like :class:OptOutError this is not a device fault, but unlike it the event is not performed: surfaced to the head-end as Table 31 status 253 (Invalid) rather than a capability-limit code.

Callers must not latch this against the device and mode. A corrected value arriving on the next event has to be allowed through -- the value is at fault, not the pairing of device and mode.

Source code in src/py20305/connectors/control_errors.py

class InvalidControlValueError(ControlError):
    """A control parameter is outside the range the IEEE 2030.5 profile allows.

    Like :class:`OptOutError` this is not a device fault, but unlike it the
    event is not performed: surfaced to the head-end as Table 31 status 253
    (Invalid) rather than a capability-limit code.

    Callers must not latch this against the device and mode. A corrected value
    arriving on the next event has to be allowed through -- the value is at
    fault, not the pairing of device and mode.
    """

py20305.connectors.translation

Pure SunSpec point-path translation connector.

Translates intermediate control parameters to SunSpec Modbus point paths. No I/O, no external dependencies. Returns dicts like: {'DERVoltVar[0].Ena': 1, 'DERVoltVar[0].Crv[1].Pt[0].V': 240}

Also includes translate_mup() for MirrorUsagePoint -> SunSpec point mapping.

Synchronous API

All translation logic is pure computation. The translate_to_sunspec() function provides synchronous access for callers that cannot use await (e.g. a threaded host application). It accepts the same (method_name, params) tuples returned by translate_controls().

ConnectorTranslation

Bases: BaseConnector

Translates intermediate parameters to SunSpec Modbus point paths.

Source code in src/py20305/connectors/translation.py

class ConnectorTranslation(BaseConnector):
    """Translates intermediate parameters to SunSpec Modbus point paths."""

    connector_name: str = "TranslationConnector"

    # ------------------------------------------------------------------
    # Curve-based modes
    # ------------------------------------------------------------------

    async def update_qv(self, params: dict[str, Any]) -> ConnectorPayload:
        return _translate_qv(params)

    async def update_pv(self, params: dict[str, Any]) -> ConnectorPayload:
        return _translate_pv(params)

    async def update_qp(self, params: dict[str, Any]) -> ConnectorPayload:
        return _translate_qp(params)

    # ------------------------------------------------------------------
    # Power limits
    # ------------------------------------------------------------------

    async def update_p_lim(self, params: dict[str, Any]) -> ConnectorPayload:
        return _translate_p_lim(params)

    # ------------------------------------------------------------------
    # Frequency droop
    # ------------------------------------------------------------------

    async def update_pf(self, params: dict[str, Any]) -> ConnectorPayload:
        return _translate_pf(params)

    # ------------------------------------------------------------------
    # Constant reactive power
    # ------------------------------------------------------------------

    async def update_const_q(self, params: dict[str, Any]) -> ConnectorPayload:
        return _translate_const_q(params)

    # ------------------------------------------------------------------
    # Constant power factor
    # ------------------------------------------------------------------

    async def update_const_pf(self, params: dict[str, Any]) -> ConnectorPayload:
        return _translate_const_pf(params)

    # ------------------------------------------------------------------
    # Fixed active power
    # ------------------------------------------------------------------

    async def update_fixed_w(self, params: dict[str, Any]) -> ConnectorPayload:
        return _translate_fixed_w(params)

    # ------------------------------------------------------------------
    # Voltage ride-through
    # ------------------------------------------------------------------

    async def update_ov(self, params: dict[str, Any]) -> ConnectorPayload:
        return _translate_ov(params)

    async def update_uv(self, params: dict[str, Any]) -> ConnectorPayload:
        return _translate_uv(params)

    async def update_ov_mc(self, params: dict[str, Any]) -> ConnectorPayload:
        return _translate_ov_mc(params)

    async def update_uv_mc(self, params: dict[str, Any]) -> ConnectorPayload:
        return _translate_uv_mc(params)

    # ------------------------------------------------------------------
    # Frequency ride-through
    # ------------------------------------------------------------------

    async def update_of(self, params: dict[str, Any]) -> ConnectorPayload:
        return _translate_of(params)

    async def update_uf(self, params: dict[str, Any]) -> ConnectorPayload:
        return _translate_uf(params)

translate_to_sunspec

translate_to_sunspec(method_name: str, params: dict[str, Any]) -> ConnectorPayload

Synchronously translate control params to SunSpec point paths.

Accepts the same (method_name, params) tuples returned by translate_controls() and translate_default_controls().

Returns a dict of SunSpec point paths, or None if the method is not a recognized translation method (e.g. methods only defined on BaseConnector with no translation logic).

Source code in src/py20305/connectors/translation.py

def translate_to_sunspec(method_name: str, params: dict[str, Any]) -> ConnectorPayload:
    """Synchronously translate control params to SunSpec point paths.

    Accepts the same ``(method_name, params)`` tuples returned by
    ``translate_controls()`` and ``translate_default_controls()``.

    Returns a dict of SunSpec point paths, or ``None`` if the method
    is not a recognized translation method (e.g. methods only defined
    on BaseConnector with no translation logic).
    """
    translator = _SYNC_TRANSLATORS.get(method_name)
    if translator is None:
        return None
    return translator(params)

translate_mup

translate_mup(mup_readings: list[dict[str, Any]]) -> dict[str, Any]

Translate MirrorUsagePoint meter readings to SunSpec points.

Parameters:

Name	Type	Description	Default
mup_readings	list[dict[str, Any]]	List of reading dicts, each with: - uom: IEEE 2030.5 unit of measure code - value: The reading value - multiplier: Power-of-ten multiplier (default 0)	required

Returns:

Type	Description
dict[str, Any]	Dict of SunSpec point paths to converted values.

Source code in src/py20305/connectors/translation.py

def translate_mup(mup_readings: list[dict[str, Any]]) -> dict[str, Any]:
    """Translate MirrorUsagePoint meter readings to SunSpec points.

    Args:
        mup_readings: List of reading dicts, each with:
            - uom: IEEE 2030.5 unit of measure code
            - value: The reading value
            - multiplier: Power-of-ten multiplier (default 0)

    Returns:
        Dict of SunSpec point paths to converted values.
    """
    points: dict[str, Any] = {}
    for reading in mup_readings:
        uom = reading.get("uom")
        if uom is None or uom not in UOM_TO_SUNSPEC:
            continue

        value = reading.get("value")
        if value is None:
            continue

        ieee_mult = reading.get("multiplier", 0) or 0
        sunspec_point, sunspec_mult = UOM_TO_SUNSPEC[uom]

        try:
            converted = int(value * (10 ** (ieee_mult - sunspec_mult)))
            points[sunspec_point] = converted
        except (TypeError, ValueError):
            continue

    return points

py20305.commands

Command vocabulary, and the observer interface that records it.

A control reaching a device is one :class:CommandRecord: what was commanded, by which interface, when, and whether the device has confirmed it. Monitoring adapters read that state so a master can see what the command interface is doing without being able to interfere, and an operator can tell why a setpoint moved.

:class:CommandObserver exists so the write path can say "this was commanded" without naming who keeps the record. The write funnels are published client code and the plane that stores the records is not, so the dependency has to point down: the funnels depend on this Protocol, and the host application injects the implementation. :class:NullCommandObserver is the default, which is what lets an embedded consumer run the same write path with no plane at all.

Reflection is deliberately not modeled here as a value the device reports. A device may clamp a setpoint, ramp toward it, or reject it, so commanded and achieved routinely differ; the record says what was asked for and how far the confirmation has got, and never pretends the two are the same.

CommandOrigin

Bases: StrEnum

The origins this package produces itself.

An origin says where a command entered the system, and it is a label: any string. Everything that takes one -- the dispatcher, :class:CommandGate, :class:CommandObserver, :class:CommandRecord -- carries it through without interpreting it. So an application that puts another interface in front of the same devices names that interface's origin itself, and passes the string. Nothing here has to learn the name first, and which interfaces an application has is not something this package should have to list.

These three are the exception, because the writes are made here: an IEEE 2030.5 control, and the two cases where the client reasserts state it already owns. Those two are not external commands at all, but they change the device just as visibly, and leaving them unrecorded is what makes a setpoint appear to move on its own. The members are strings, so a gate or an observer compares and stores them like any other origin.

Source code in src/py20305/commands.py

class CommandOrigin(StrEnum):
    """The origins this package produces itself.

    An origin says where a command entered the system, and it is a label: any
    string. Everything that takes one -- the dispatcher, :class:`CommandGate`,
    :class:`CommandObserver`, :class:`CommandRecord` -- carries it through
    without interpreting it. So an application that puts another interface in
    front of the same devices names that interface's origin itself, and passes
    the string. Nothing here has to learn the name first, and which interfaces
    an application has is not something this package should have to list.

    These three are the exception, because the writes are made here: an IEEE
    2030.5 control, and the two cases where the client reasserts state it
    already owns. Those two are not external commands at all, but they change
    the device just as visibly, and leaving them unrecorded is what makes a
    setpoint appear to move on its own. The members are strings, so a gate or
    an observer compares and stores them like any other origin.
    """

    IEEE2030_5 = "ieee2030_5"
    #: DdercTracker reapplying DefaultDERControl on the fallback path after an
    #: event completes. The classic "my setpoint reverted and nothing says why".
    DDERC_REAPPLY = "dderc_reapply"
    #: Loss-of-communications handling reasserting the planning limit.
    COMMS_LOSS = "comms_loss"
    #: An event ending: the modes it set that the DefaultDERControl does not
    #: carry are disabled, so a device does not keep a setpoint whose event
    #: is over.
    EVENT_END = "event_end"

CommandStatus

Bases: StrEnum

How far a command has got.

UNCONFIRMED is the honest default and often the terminal state: the connector contract exposes no general setpoint readback, so most controls have nothing to confirm against. It means "we asked and the write returned without error", not "the device is doing it".

Source code in src/py20305/commands.py

class CommandStatus(StrEnum):
    """How far a command has got.

    ``UNCONFIRMED`` is the honest default and often the terminal state: the
    connector contract exposes no general setpoint readback, so most controls
    have nothing to confirm against. It means "we asked and the write returned
    without error", not "the device is doing it".
    """

    UNCONFIRMED = "unconfirmed"
    CONFIRMED = "confirmed"
    REJECTED = "rejected"

CommandRecord dataclass

One control applied to one device.

Source code in src/py20305/commands.py

@dataclass(frozen=True, slots=True)
class CommandRecord:
    """One control applied to one device."""

    #: The connector operation, e.g. ``p_lim`` or ``es_permit_service``. Keyed on
    #: the operation rather than a measured quantity: a limit and the power that
    #: results from it are different things (see the module docstring).
    control: str
    #: What was sent, verbatim. Curve modes carry point lists here; nothing in
    #: this layer interprets the shape.
    params: Mapping[str, Any]
    #: Where the command entered: one of :class:`CommandOrigin`, or whatever
    #: the application calls the interface that issued it.
    origin: str
    #: Wall-clock epoch seconds at which the write was issued. Compared against
    #: an observation's read-start time to decide whether that observation could
    #: possibly reflect this command.
    commanded_at: float
    status: CommandStatus
    confirmed_at: float | None = None
    #: What the connector raised, when status is REJECTED.
    error: str | None = None

CommandObserver

Bases: Protocol

Where the write path reports what it did.

Two operations, because confirmation arrives separately from the command and usually from a different code path: the write funnel knows what was asked for, and whoever polls the device later knows what it says about itself.

Source code in src/py20305/commands.py

@runtime_checkable
class CommandObserver(Protocol):
    """Where the write path reports what it did.

    Two operations, because confirmation arrives separately from the command and
    usually from a different code path: the write funnel knows what was asked
    for, and whoever polls the device later knows what it says about itself.
    """

    def record_command(
        self,
        device: str,
        control: str,
        params: Mapping[str, Any],
        *,
        origin: str,
        at: float,
        error: str | None = None,
    ) -> None:
        """Record that ``control`` was applied to ``device``, or failed to be."""
        ...

    def record_readback(
        self, device: str, observed: Mapping[str, Any], *, read_started_at: float
    ) -> None:
        """Offer a device's self-reported state as confirmation evidence.

        ``read_started_at`` is when the read that produced ``observed`` *began*,
        not when it completed. An observation that started before a command was
        issued cannot reflect it, however recent its arrival, and is what the
        implementation uses to keep an in-flight poll from clearing a command it
        never saw.
        """
        ...

record_command

record_command(device: str, control: str, params: Mapping[str, Any], *, origin: str, at: float, error: str | None = None) -> None

Record that control was applied to device, or failed to be.

Source code in src/py20305/commands.py

def record_command(
    self,
    device: str,
    control: str,
    params: Mapping[str, Any],
    *,
    origin: str,
    at: float,
    error: str | None = None,
) -> None:
    """Record that ``control`` was applied to ``device``, or failed to be."""
    ...

record_readback

record_readback(device: str, observed: Mapping[str, Any], *, read_started_at: float) -> None

Offer a device's self-reported state as confirmation evidence.

read_started_at is when the read that produced observed began, not when it completed. An observation that started before a command was issued cannot reflect it, however recent its arrival, and is what the implementation uses to keep an in-flight poll from clearing a command it never saw.

Source code in src/py20305/commands.py

def record_readback(
    self, device: str, observed: Mapping[str, Any], *, read_started_at: float
) -> None:
    """Offer a device's self-reported state as confirmation evidence.

    ``read_started_at`` is when the read that produced ``observed`` *began*,
    not when it completed. An observation that started before a command was
    issued cannot reflect it, however recent its arrival, and is what the
    implementation uses to keep an in-flight poll from clearing a command it
    never saw.
    """
    ...

NullCommandObserver

Discards everything. The default where no plane is wired.

Published write paths run unchanged in a deployment that keeps no commanded state -- an embedded single-upstream consumer, or any test that cares about the write rather than the record.

Source code in src/py20305/commands.py

class NullCommandObserver:
    """Discards everything. The default where no plane is wired.

    Published write paths run unchanged in a deployment that keeps no commanded
    state -- an embedded single-upstream consumer, or any test that cares about
    the write rather than the record.
    """

    def record_command(
        self,
        device: str,
        control: str,
        params: Mapping[str, Any],
        *,
        origin: str,
        at: float,
        error: str | None = None,
    ) -> None:
        """No-op."""

    def record_readback(
        self, device: str, observed: Mapping[str, Any], *, read_started_at: float
    ) -> None:
        """No-op."""

record_command

record_command(device: str, control: str, params: Mapping[str, Any], *, origin: str, at: float, error: str | None = None) -> None

No-op.

Source code in src/py20305/commands.py

def record_command(
    self,
    device: str,
    control: str,
    params: Mapping[str, Any],
    *,
    origin: str,
    at: float,
    error: str | None = None,
) -> None:
    """No-op."""

record_readback

record_readback(device: str, observed: Mapping[str, Any], *, read_started_at: float) -> None

No-op.

Source code in src/py20305/commands.py

def record_readback(
    self, device: str, observed: Mapping[str, Any], *, read_started_at: float
) -> None:
    """No-op."""

CommandGate

Bases: Protocol

Whether an origin may command a device.

A deployment may serve several command interfaces while intending only one of them to command any given device. Deciding that in configuration alone leaves it unenforced at the write: every interface still reaches the dispatcher, so one interface's events and another's setpoints overwrite each other while the configuration reads as though a single one were in charge.

Injected rather than imported, for the same reason :class:CommandObserver is. Which interface holds authority over a device is the consuming application's concern, and the dispatcher is client code, so the dependency points that way.

Source code in src/py20305/commands.py

@runtime_checkable
class CommandGate(Protocol):
    """Whether an origin may command a device.

    A deployment may serve several command interfaces while intending only one of
    them to command any given device. Deciding that in configuration alone leaves
    it unenforced at the write: every interface still reaches the dispatcher, so
    one interface's events and another's setpoints overwrite each other while the
    configuration reads as though a single one were in charge.

    Injected rather than imported, for the same reason :class:`CommandObserver`
    is. Which interface holds authority over a device is the consuming
    application's concern, and the dispatcher is client code, so the dependency
    points that way.
    """

    def may_command(self, device: str, origin: str) -> bool:
        """Whether ``origin`` holds the command role for ``device``.

        Deny an origin you do not recognize. An origin is a string the caller
        names, so nothing upstream of this method rejects one that no interface
        owns: a mistyped name, or a write path added without one, arrives here
        looking like any other. An implementation that answers only for the
        origins it knows, and refuses the rest, is what makes that a refused
        write. One that permits by default makes it a command.
        """
        ...

may_command

may_command(device: str, origin: str) -> bool

Whether origin holds the command role for device.

Deny an origin you do not recognize. An origin is a string the caller names, so nothing upstream of this method rejects one that no interface owns: a mistyped name, or a write path added without one, arrives here looking like any other. An implementation that answers only for the origins it knows, and refuses the rest, is what makes that a refused write. One that permits by default makes it a command.

Source code in src/py20305/commands.py

def may_command(self, device: str, origin: str) -> bool:
    """Whether ``origin`` holds the command role for ``device``.

    Deny an origin you do not recognize. An origin is a string the caller
    names, so nothing upstream of this method rejects one that no interface
    owns: a mistyped name, or a write path added without one, arrives here
    looking like any other. An implementation that answers only for the
    origins it knows, and refuses the rest, is what makes that a refused
    write. One that permits by default makes it a command.
    """
    ...

AllowAllCommands

Permits everything. The default where no authority is wired.

A consumer with a single command interface, or a test concerned with the write rather than with who was allowed to make it, behaves exactly as it did before the gate existed.

Everything includes an origin nothing claims. With this gate a mistyped origin commands like a correct one, and the only trace is the name in the audit record. That is the right trade for one interface, where there is no second commander for a wrong name to be mistaken for. An application with more than one should wire a gate of its own, and leaving this one in place is then a decision to check nothing.

Source code in src/py20305/commands.py

class AllowAllCommands:
    """Permits everything. The default where no authority is wired.

    A consumer with a single command interface, or a test concerned with the
    write rather than with who was allowed to make it, behaves exactly as it did
    before the gate existed.

    Everything includes an origin nothing claims. With this gate a mistyped
    origin commands like a correct one, and the only trace is the name in the
    audit record. That is the right trade for one interface, where there is no
    second commander for a wrong name to be mistaken for. An application with
    more than one should wire a gate of its own, and leaving this one in place
    is then a decision to check nothing.
    """

    def may_command(self, device: str, origin: str) -> bool:
        """Always True."""
        return True

may_command

may_command(device: str, origin: str) -> bool

Always True.

Source code in src/py20305/commands.py

def may_command(self, device: str, origin: str) -> bool:
    """Always True."""
    return True

CommandNotPermittedError

Bases: Exception

A gate refused this origin authority over this device.

Distinct from a connector error because the cause and the remedy are different: nothing is wrong with the device or the transport, and the caller cannot retry its way out of it. A caller serving a protocol client needs the distinction to answer that client honestly -- refused is not the same answer as failed.

Raised only by the entry points whose caller has somewhere to put it. A control arriving from an event engine is reported and dropped instead, since an interface posting to a device another one commands is a configuration being honored rather than a fault.

Source code in src/py20305/commands.py

class CommandNotPermittedError(Exception):
    """A gate refused this origin authority over this device.

    Distinct from a connector error because the cause and the remedy are
    different: nothing is wrong with the device or the transport, and the caller
    cannot retry its way out of it. A caller serving a protocol client needs the
    distinction to answer that client honestly -- refused is not the same answer
    as failed.

    Raised only by the entry points whose caller has somewhere to put it. A
    control arriving from an event engine is reported and dropped instead, since
    an interface posting to a device another one commands is a configuration
    being honored rather than a fault.
    """