Client

The protocol client: discovery, polling, mTLS transport and retry.

py20305.client.csip_client

High-level CSIP client with lifecycle, polling, and 404 recovery.

CsipClient

Async IEEE 2030.5 CSIP client with discovery, polling, and shutdown.

Source code in src/py20305/client/csip_client.py

class CsipClient:
    """Async IEEE 2030.5 CSIP client with discovery, polling, and shutdown."""

    def __init__(
        self,
        base_url: str,
        *,
        tls: TlsConfig | None = None,
        retry: RetryPolicy | None = None,
        dispatcher: ControlDispatcher | None = None,
        subscription_manager: SubscriptionManager | None = None,
        notification_server: NotificationServer | None = None,
        on_structural_change: Callable[[], Awaitable[None]] | None = None,
        on_device_removed: Callable[[str], Awaitable[None]] | None = None,
        server_2018_compat: bool = False,
        always_send_alarm_status: bool = False,
        request_headers: dict[str, str] | None = None,
        dcap_path: str = "/dcap",
        heartbeat_enabled: bool = True,
        connection_heartbeat_seconds: int = 120,
        reconcile_enabled: bool = True,
        reconcile_interval_seconds: int = 0,
        renewal_interval_seconds: int = 86400,
        group_lookup: Callable[[str], list[str] | None] | None = None,
        registration_pins: dict[str, int] | None = None,
        comms_loss_seconds: int = 0,
        comms_loss_eval_seconds: int = 30,
        use_server_time: bool = True,
        time_drift_warn_seconds: int = 30,
        fsa_stale_seconds: float | None = None,
        pricing_enabled: bool = False,
    ) -> None:
        #: Application-level server timebase :
        #: time-of-day-sensitive operations follow the head-end's Time resource;
        #: the OS clock is never touched. Shared via Sep2Client (discovery,
        #: telemetry) and EventProcessor.
        self._timebase = ServerTimebase(
            enabled=use_server_time,
            drift_warn_seconds=time_drift_warn_seconds,
            fsa_stale_seconds=fsa_stale_seconds,
        )
        self._http = Sep2Client(
            base_url,
            tls=tls,
            retry=retry,
            server_2018_compat=server_2018_compat,
            always_send_alarm_status=always_send_alarm_status,
            request_headers=request_headers,
            timebase=self._timebase,
        )
        self._state = DiscoveredState()
        #: Pricing function set opt-in (config-driven). Preserved across
        #: rediscovery (state.clear() keeps it) and gates tariff discovery/poll.
        self._state.pricing_enabled = pricing_enabled
        self._heartbeat_enabled = heartbeat_enabled
        self._connection_heartbeat_seconds = connection_heartbeat_seconds
        self._reconcile_enabled = reconcile_enabled
        self._reconcile_interval_seconds = reconcile_interval_seconds
        self._renewal_interval_seconds = renewal_interval_seconds
        self._scheduler = PollScheduler(heartbeat_enabled=heartbeat_enabled)
        self._shutdown_event = asyncio.Event()
        self._rediscovery_lock = asyncio.Lock()
        self._state_ready = asyncio.Event()
        self._state_ready.set()  # state is valid initially
        self._dispatcher: ControlDispatcher = dispatcher or NullDispatcher()
        #: Loss-of-communications mode. Shared with the EventProcessor so the
        #: time-based detector here can flip it and the processor gates control
        #: application on it. Zero seconds disables detection entirely.
        if comms_loss_seconds > 0 and comms_loss_eval_seconds <= 0:
            # A non-positive cadence would make the scheduler busy-loop
            # (wait_for timeout of 0). The configuration model enforces ge=1; this
            # guards direct constructions.
            raise ValueError(
                "comms_loss_eval_seconds must be positive when comms-loss "
                "detection is enabled (comms_loss_seconds > 0)"
            )
        self._comms_loss_seconds = comms_loss_seconds
        self._comms_loss_eval_seconds = comms_loss_eval_seconds
        self._comms_loss = CommsLossState()
        #: Expiry timer for an operator-triggered outage simulation. A window is
        #: never open-ended: a forgotten one would strand a live site at the
        #: planning limit.
        self._comm_loss_simulation_expiry: asyncio.Task[None] | None = None
        #: Serializes simulate/clear/expiry. Without it a timer that has already
        #: begun clearing can finish against a window someone else has since
        #: opened, marking a live simulation recovered.
        self._comm_loss_lifecycle = asyncio.Lock()
        #: Incremented per window, so an older timer can tell it woke too late.
        self._comm_loss_generation = 0
        #: The client's own certificate LFDI, captured at connect() for in-band
        #: self-reregistration on comms-loss recovery. Cert-derived and stable.
        self._own_lfdi: str | None = None
        #: Audit trail for event lifecycle and loss-of-communications records.
        #: Disabled until the host attaches a forwarder and configures it.
        self._audit = AuditEmitter()
        #: Whether the current comms-loss episode was a simulation, so the
        #: clear record attributes it the same way the entry record did.
        self._comms_loss_simulated = False
        #: Monotonic time the current comms-loss episode began, for how long it
        #: lasted. Monotonic because the server timebase can move under it.
        self._comms_loss_entered_monotonic: float | None = None
        self._event_processor = EventProcessor(
            self._http,
            self._state,
            self._dispatcher,
            self._shutdown_event,
            state_ready=self._state_ready,
            group_lookup=group_lookup,
            comms_loss=self._comms_loss,
            timebase=self._timebase,
            audit=self._audit,
        )
        #: Pricing function set: relays active-interval prices to connectors.
        #: Constructed unconditionally but only driven by the tariff poll when
        #: pricing is enabled (see _start_polls).
        self._tariff_processor = TariffProcessor(
            self._http, self._state, self._dispatcher, timebase=self._timebase
        )
        self._subscription_manager = subscription_manager
        self._notification_server = notification_server
        self._on_structural_change = on_structural_change
        self._on_rediscovered: Callable[[], Awaitable[None]] | None = None
        self._on_device_removed = on_device_removed
        self._dcap_path = dcap_path
        #: LFDI hex -> expected registration PIN. Verified once during the
        #: initial connect() discovery (not on rediscovery). Empty/None
        #: disables the check.
        self._registration_pins = registration_pins or None
        self._renewal_task: asyncio.Task[None] | None = None
        #: Gap 4: Track consecutive poll failures for connectivity recovery
        self._poll_failure_count = 0
        #: Set when trigger_rediscovery() is called while rediscovery is already running.
        #: The running rediscovery will loop once more after completing.
        self._rediscovery_pending = False
        #: One-shot guard for the "no SubscriptionListLink anywhere" warning.
        #: ``_auto_subscribe`` is called on every poll (default 900s) and after every
        #: rediscovery; without this guard a server that never exposes the link would
        #: emit a WARNING ~96 times/day. Reset to False once a link is discovered so a
        #: subsequent server-side fix is also logged (and a regression re-warns).
        self._no_sub_link_warned = False

    @property
    def dispatcher(self) -> ControlDispatcher:
        """The dispatcher this client applies controls through."""
        return self._dispatcher

    @property
    def audit(self) -> AuditEmitter:
        """The audit trail emitter. Attach a forwarder and configure it to enable."""
        return self._audit

    @property
    def subscription_manager(self) -> SubscriptionManager | None:
        """The subscription manager, when subscriptions are wired."""
        return self._subscription_manager

    def attach_subscriptions(
        self,
        manager: SubscriptionManager,
        notification_server: NotificationServer | None = None,
    ) -> None:
        """Wire subscribe/notify into this client after construction.

        The manager needs this client's transport, which exists only once the
        client does, so a caller building both cannot pass them to
        ``__init__`` -- construction order forces attachment to come second.
        Attach before ``connect()``: auto-subscription and poll suppression
        consult the manager during connection.
        """
        self._subscription_manager = manager
        if notification_server is not None:
            self._notification_server = notification_server
            notification_server.on_notification = self._handle_notification

    def set_on_structural_change(
        self, callback: Callable[[], Awaitable[None]] | None
    ) -> None:
        """Set the structural-change callback after construction.

        The same construction-order problem as ``attach_subscriptions``, one
        step later: telemetry is started once discovery has produced the hrefs
        it posts to, so what wants to hear about a later structural change does
        not exist when this client is built.

        Note what this callback replaces. Where a structural notification would
        otherwise call ``trigger_rediscovery()``, it calls this instead, so a
        callback that does not rediscover leaves the state it was called about
        unrefreshed. To act on rebuilt state, use ``set_on_rediscovered``.
        """
        self._on_structural_change = callback

    def set_on_rediscovered(self, callback: Callable[[], Awaitable[None]] | None) -> None:
        """Set a callback to run after any rediscovery pass that completed.

        Every path that rebuilds discovered state ends here -- a structural
        notification, 404 recovery, comms-loss recovery, a subscription lost,
        an operator's request through the management API -- so anything holding
        hrefs from the previous state refreshes once, in one place, rather than
        each trigger remembering to say so.

        Not called when discovery raised, because the state was not rebuilt,
        nor when the call coalesced into a running rediscovery, because that
        one calls it on completion.
        """
        self._on_rediscovered = callback

    @property
    def http(self) -> Sep2Client:
        return self._http

    @property
    def state(self) -> DiscoveredState:
        return self._state

    @property
    def timebase(self) -> ServerTimebase:
        return self._timebase

    async def run_redirect_probe(self) -> dict[str, Any]:
        """Run ERR-001 HTTP-to-HTTPS redirect probe using this client's session."""
        from py20305.client.redirect_probe import run_redirect_probe

        return await run_redirect_probe(self._http.host, self._http)

    async def register_end_device(
        self,
        *,
        lfdi: str,
        device_category: int | None = None,
        edev_list_href: str | None = None,
        check_duplicate: bool = True,
    ) -> str | None:
        """In-band registration: POST a managed EndDevice to the server's
        EndDeviceList and return the ``Location`` (the new ``/edev/{id}`` href).

        ``lfdi`` is a 40-hex string; ``sFDI`` is derived from it (never
        free-entered). ``device_category`` is an optional DeviceCategoryType
        bitmap. The target list defaults to the discovered
        ``dcap.end_device_list_link`` unless ``edev_list_href`` overrides it.

        Raises ``ValueError`` for a missing EndDeviceList href or, when
        ``check_duplicate``, an LFDI the server already lists (IEEE 2030.5 §8.5.3:
        clients SHALL NOT POST a duplicate EndDevice). Raises ``Sep2ProtocolError``
        on a server rejection status.
        """
        from py20305.models.sep.sep import (
            DeviceCategoryType,
            EndDevice,
            Sfditype,
            TimeType,
        )
        from py20305.security.identity import compute_sfdi

        lfdi_bytes = bytes.fromhex(lfdi)
        href = edev_list_href or _extract_href(
            self._state.dcap.end_device_list_link if self._state.dcap else None
        )
        if not href:
            raise ValueError(
                "No EndDeviceList href -- the server's DeviceCapability has not been discovered"
            )

        if check_duplicate and await self._server_lists_end_device(lfdi, edev_list_href=href):
            raise ValueError(f"An EndDevice with LFDI {lfdi} is already registered")

        end_device = EndDevice(
            l_fdi=lfdi_bytes,
            s_fdi=Sfditype(value=compute_sfdi(lfdi)),
            changed_time=TimeType(value=int(self._timebase.now())),
        )
        if device_category is not None:
            if not 0 <= device_category < 2**32:
                raise ValueError("device_category must be between 0 and 4294967295 (32-bit bitmap)")
            end_device.device_category = DeviceCategoryType(
                value=device_category.to_bytes(4, "big")
            )
        return await self._http.post(href, end_device)

    async def _server_lists_end_device(self, lfdi: str, edev_list_href: str | None = None) -> bool:
        """Whether the server's EndDeviceList currently contains ``lfdi``.

        Live paged GET against the discovered EndDeviceList (or
        ``edev_list_href``). Raises ``ValueError`` when no EndDeviceList href is
        known. Used for the §8.5.3 duplicate check in ``register_end_device``
        and to make comms-loss recovery reregistration conditional: some servers
        (e.g. certain utility head-ends) drop EndDevices during an outage and
        expect the client to re-register, while others retain them and must not
        receive a duplicate POST.
        """
        from py20305.models.sep.sep import EndDeviceList

        href = edev_list_href or _extract_href(
            self._state.dcap.end_device_list_link if self._state.dcap else None
        )
        if not href:
            raise ValueError(
                "No EndDeviceList href -- the server's DeviceCapability has not been discovered"
            )
        lfdi_bytes = bytes.fromhex(lfdi)
        for page in await self._http.get_list(href, EndDeviceList):
            for existing in page.end_device or []:
                if existing.l_fdi == lfdi_bytes:
                    return True
        return False

    async def connect(self) -> None:
        """Run initial resource discovery and auto-subscribe if configured."""
        if self._notification_server and not self._notification_server.running:
            await self._notification_server.start()
        await discover(
            self._http,
            self._state,
            registration_pins=self._registration_pins,
            dcap_path=self._dcap_path,
        )
        # Capture the client's own LFDI for comms-loss self-reregistration.
        # Cert-derived and stable across reset_session, so a single capture holds.
        if self._own_lfdi is None:
            self._own_lfdi = self._http.client_lfdi
        if self._subscription_manager:
            if self._subscription_manager.active_subscriptions:
                valid, removed = await self._subscription_manager.validate_restored_subscriptions()
                logger.info("Subscription validation: %d valid, %d removed", valid, removed)
            await self._auto_subscribe()
            self._update_poll_suppression()
            if self._subscription_manager.active_subscriptions:
                self._renewal_task = asyncio.create_task(
                    self._subscription_manager.start_renewal_task(
                        self._shutdown_event, interval_seconds=self._renewal_interval_seconds
                    )
                )

    async def run(self) -> None:
        """Start polling and block until shutdown is signaled."""
        logger.info("Starting poll loop")
        # Process controls from initial discovery immediately
        for href in list(self._state.der_programs):
            await self._event_processor.process_controls(href)
        self._start_polls()
        await self._shutdown_event.wait()
        logger.info("Poll loop exiting (shutdown signaled)")

    def _start_polls(self) -> None:
        rates = self._state.poll_rates
        logger.debug("Scheduling polls: %s", {k: v for k, v in rates.items() if v is not None})

        dcap_rate = rates.get("dcap")
        if dcap_rate is not None:
            self._scheduler.schedule("dcap", dcap_rate, self._poll_dcap)

        edev_rate = rates.get("edev")
        if edev_rate is not None:
            self._scheduler.schedule("edev", edev_rate, self._poll_edev)

        fsa_rate = rates.get("fsa")
        if fsa_rate is not None:
            self._scheduler.schedule("fsa", fsa_rate, self._poll_fsa)

        derp_rate = rates.get("derp")
        if derp_rate is not None:
            self._scheduler.schedule("derp", derp_rate, self._poll_derp)

        # Pricing function set (opt-in). Uses the tariff pollRate captured in
        # discovery, or the default when a tariff tree hasn't been seen yet (so
        # tariffs that appear later are still picked up).
        if self._state.pricing_enabled:
            tariff_rate = rates.get("tariff") or DEFAULT_POLL_RATE
            self._scheduler.schedule("tariff", tariff_rate, self._poll_tariff)

        time_rate = rates.get("time")
        if time_rate is not None:
            # The staleness threshold has to know this: a per-FSA observation
            # is at its oldest just before the next successful refresh, so a
            # threshold shorter than the poll interval retires healthy scopes.
            self._timebase.note_time_poll_rate(time_rate)
            self._scheduler.schedule("time", time_rate, self._poll_time)

        # Connectivity liveness heartbeat (issue: stale server_alive). Runs on a
        # short, fixed cadence independent of resource poll rates so server_alive
        # / last_contact_epoch can't go stale-true while only telemetry PUT/POSTs
        # are in flight. Never subscription-suppressed; idle-gated in the probe.
        if self._connection_heartbeat_seconds > 0:
            self._scheduler.schedule(
                "connectivity", self._connection_heartbeat_seconds, self._connectivity_probe
            )

        # Loss-of-communications detector.
        # Time-based: enters when last_contact_epoch is older than the window,
        # recovers on the first fresh contact. The connectivity heartbeat above
        # keeps issuing GETs during the outage so contact refreshes on restore.
        # Re-scheduled on every rediscovery (_start_polls re-runs), so it
        # self-rearms.
        if self._comms_loss_seconds > 0:
            self._scheduler.schedule(
                "comms_loss", self._comms_loss_eval_seconds, self._comms_loss_probe
            )

        # SubscriptionList reconcile (issue: server drops a subscription after a
        # notification-delivery failure, and we never notice). Polls the server's
        # SubscriptionList and re-establishes any locally-active subscription it's
        # missing. Runs only when subscriptions exist; never subscription-
        # suppressed (the SubscriptionList itself isn't subscribed -- rule (r)).
        if self._reconcile_enabled and self._subscription_manager is not None:
            sub_rate = self._sub_poll_rate()
            if sub_rate is not None:
                self._scheduler.schedule("sub", sub_rate, self._poll_sub_reconcile)

    def _sub_poll_rate(self) -> int | None:
        """Cadence for the SubscriptionList reconcile poll, in seconds.

        Follows the server-advertised SubscriptionList ``pollRate`` (captured into
        ``poll_rates["sub"]`` at discovery, default 900s) unless
        ``reconcile_interval_seconds`` is set (>0), which forces a fixed override.
        The value is clamped to the same safe lower bound as other poll rates.
        """
        override = self._reconcile_interval_seconds
        raw_rate = override if override > 0 else self._state.poll_rates.get("sub")
        return normalize_poll_rate(raw_rate, resource_key="sub", default=DEFAULT_POLL_RATE)

    async def _poll_sub_reconcile(self) -> None:
        """Reconcile local subscriptions against the server's SubscriptionList."""
        if self._subscription_manager is None:
            return
        agg_sub_href = self._find_agg_sub_href()
        if agg_sub_href is None:
            return  # no SubscriptionList discovered yet; nothing to reconcile
        if not self._subscription_manager.active_subscriptions:
            return  # nothing to reconcile
        await self._subscription_manager.reconcile_with_server(agg_sub_href)

    async def _connectivity_probe(self) -> None:
        """Cheap liveness GET, but only when the connection isn't confirmed alive.

        The IEEE chain audit now runs at the TLS handshake, so *any* successful
        request -- a GET or a telemetry PUT/POST -- both sets ``server_alive`` True
        and refreshes ``last_validated_epoch``. So the gate is: skip only when the
        connection is currently alive *and* a request confirmed it within the
        heartbeat window. The probe still fires while ``server_alive`` is False, or
        when nothing has kept the connection warm within the window (e.g. a
        poll-suppressing subscription with only sparse traffic) -- issuing a small
        GET (Time, falling back to DeviceCapability) whose outcome refreshes
        ``server_alive`` via the normal request path. Telemetry reviving
        ``server_alive`` directly is what removes the old stuck-false-after-
        reconnect failure mode. Errors are expected when the server is down and are
        left to the poll loop's handler.
        """
        from py20305.models.sep.sep import DeviceCapability, Time

        if self._http.server_alive:
            last_validated = self._http.last_validated_epoch
            if (
                last_validated is not None
                and (time.time() - last_validated) < self._connection_heartbeat_seconds
            ):
                return  # alive and a validating GET confirmed it within the window

        if self._state.time_href:
            # Free timebase refresh: the probe already GETs Time, so feed the
            # response into the shared timebase instead of discarding it.
            t = await self._http.get(self._state.time_href, Time)
            observe_time_resource(self._timebase, t, href=self._state.time_href)
        else:
            await self._http.get(self._dcap_path, DeviceCapability)

    async def _comms_loss_probe(self) -> None:
        """Detect and recover from loss of communications (time-based).

        Enters comms-loss mode when the last reachable contact is older than
        ``comms_loss_seconds``; recovers on the first contact fresher than that.
        Never fires before the first contact (``last_contact_epoch is None``),
        avoiding cold-start false positives. Owns both the enter and exit edges,
        independent of the count-based server-restart detection in
        ``_note_successful_contact``.
        """
        # Boundary epochs are server-derived (event ends), so compare against
        # the server timebase; the silence detection below stays on the local
        # clock (elapsed, drift-immune). Integer comparison (like the recovery
        # clear) so a fractional now() can't drop the gate up to ~1s early.
        boundary = self._comms_loss.resume_after_epoch
        if (
            not self._comms_loss.active
            and boundary is not None
            and int(self._timebase.now()) > boundary
        ):
            # The opted-out window has fully elapsed: no event starting at/before
            # the boundary can still be pending, so the resume-after gate is done.
            self._comms_loss.resume_after_epoch = None
            logger.info("Comms-loss resume boundary passed (epoch %d); cleared", boundary)
        lce = self._http.last_contact_epoch
        if lce is None:
            return
        elapsed = int(time.time() - lce)
        if not self._comms_loss.active:
            if elapsed >= self._comms_loss_seconds:
                await self._enter_comms_loss(elapsed)
        elif elapsed < self._comms_loss_seconds:
            await self._recover_from_comms_loss()

    async def _enter_comms_loss(self, elapsed: int) -> None:
        """Flag comms-loss mode and opt out of active events (idempotent)."""
        if self._comms_loss.active:
            return
        self._comms_loss.active = True
        from py20305.diagnostics import report

        # Attribution is causal, not "was a window open at the time": a link
        # already down when the operator armed a simulation stays reported as
        # the genuine outage it is, and a real fault during a window is never
        # relabelled a test artifact and hidden.
        simulated = self._http.silence_is_simulated is True
        details: dict[str, Any] = {
            "elapsed_seconds": elapsed,
            "threshold": self._comms_loss_seconds,
        }
        if simulated:
            details["simulated"] = True
        report(
            "warnings",
            f"Loss of communications: no upstream contact for {elapsed}s "
            f"(>= {self._comms_loss_seconds}s). Opting out of active events and "
            "managing the DER at the planning limit (DefaultDERControl)."
            + (" Cause: operator-triggered simulation." if simulated else ""),
            source="client",
            # Separate dedup identity per mode. The diagnostics store keeps the
            # first entry's message and details for a repeated key, so sharing
            # one key would let a simulated outage's `simulated: true` stick to
            # every genuine outage that followed it, and the reverse.
            dedup_key="comms_loss:entered:simulated" if simulated else "comms_loss:entered",
            details=details,
        )
        logger.warning("Entering loss-of-communications mode (silent for %ds)", elapsed)
        self._comms_loss_simulated = simulated
        self._comms_loss_entered_monotonic = time.monotonic()
        # Recorded before any event is opted out, so the record that explains
        # the opt-outs and the fallback writes comes ahead of them in the trail,
        # and is not lost if reverting the fleet fails part way.
        self._audit.comms_loss(
            transition="entered",
            elapsed_seconds=elapsed,
            threshold=self._comms_loss_seconds,
            simulated=simulated,
            at=self._timebase.now(),
        )
        await self._event_processor.enter_comms_loss()

    def _comms_loss_duration(self) -> int:
        """Whole seconds in loss-of-communications mode so far, or 0 if unknown."""
        started = self._comms_loss_entered_monotonic
        return int(time.monotonic() - started) if started is not None else 0

    async def _recover_from_comms_loss(self) -> None:
        """Recover once communications are restored.

        Reregisters the client's own EndDevice in-band, re-polls the latest
        schedules (superseding prior ones), resumes events after the last
        opted-out event via the resume-after boundary, then clears the mode.
        """
        logger.info("Communications restored -- recovering from loss-of-communications mode")
        # Before the re-poll, whose records it explains. Repeats if this
        # recovery fails and the next probe tick tries again.
        self._audit.comms_loss(
            transition="recovering",
            duration_seconds=self._comms_loss_duration(),
            threshold=self._comms_loss_seconds,
            simulated=self._comms_loss_simulated,
            at=self._timebase.now(),
        )
        # 1. In-band self-reregistration, only when the server no longer
        #    lists our EndDevice: some head-ends remove EndDevices during an
        #    outage and expect re-registration, while others retain them and
        #    must not receive a duplicate POST (IEEE 2030.5 §8.5.3).
        #    Best-effort: a failure is logged and recovery continues (the
        #    re-poll below is what actually resumes control). The client method
        #    (unlike the web layer) does not refuse the own LFDI;
        #    check_duplicate=False avoids re-fetching the list we just checked.
        if self._own_lfdi is not None:
            try:
                if await self._server_lists_end_device(self._own_lfdi):
                    logger.info(
                        "Comms-loss recovery: own EndDevice still listed by the "
                        "server; skipping in-band reregistration"
                    )
                else:
                    await self.register_end_device(lfdi=self._own_lfdi, check_duplicate=False)
            except (Sep2Error, ValueError) as exc:
                logger.warning("Comms-loss recovery: self-reregistration failed: %s", exc)
        # 2. Re-poll schedules, supersede, rebuild timers, manage idle devices per
        #    DDERC. The resume-after boundary makes process_controls skip the
        #    events opted out during the outage.
        if not await self.trigger_rediscovery():
            # Rediscovery raised (or coalesced): the schedule wasn't rebuilt. Stay
            # in comms-loss mode so scheduled events remain opted out, and let the
            # next probe tick retry the whole recovery (reregister + re-poll). The
            # boundary must stay set for that retry.
            logger.warning(
                "Comms-loss recovery: rediscovery did not complete; staying in "
                "loss-of-communications mode, will retry on the next probe tick"
            )
            return
        # 3. Clear the mode now the fresh schedule has been processed. The
        #    resume-after boundary is NOT cleared while it is still in the
        #    future: recovery's re-poll skips (does not store) opted-out-window
        #    events, so a later routine poll would re-classify them as new and
        #    late-apply them if the gate dropped early. The probe clears the
        #    boundary once it has passed. Opted-out event records likewise keep
        #    their opted_out flag until they prune (their end is at/before the
        #    boundary): they must not resume and must stay excluded from
        #    "other active" DDERC checks.
        self._comms_loss.active = False
        boundary = self._comms_loss.resume_after_epoch
        if boundary is not None and int(self._timebase.now()) > boundary:
            self._comms_loss.resume_after_epoch = None
        self._audit.comms_loss(
            transition="cleared",
            duration_seconds=self._comms_loss_duration(),
            threshold=self._comms_loss_seconds,
            simulated=self._comms_loss_simulated,
            at=self._timebase.now(),
            resume_after=self._comms_loss.resume_after_epoch,
        )
        self._comms_loss_entered_monotonic = None
        logger.info(
            "Loss-of-communications mode cleared%s",
            (
                f" (resume-after boundary {boundary} retained until it passes)"
                if self._comms_loss.resume_after_epoch is not None
                else ""
            ),
        )

    # ------------------------------------------------------------------
    # Operator-triggered comms-loss simulation
    # ------------------------------------------------------------------

    async def simulate_comm_loss(
        self, duration_seconds: float, *, drain_timeout: float = 5.0
    ) -> dict[str, Any]:
        """Cut this client off from the head-end for ``duration_seconds``.

        Mechanism, not policy: whether a simulation is permitted at all, and how
        long a window may reasonably run, are the caller's decisions. This
        method does what it is told and guarantees only that the window closes.

        Notifications are suspended before the transport gate closes, so a
        control cannot slip in through the inbound path while the outbound one
        is being shut. Both then drain: a request or handler already running
        would otherwise finish after the window opened, refreshing
        ``last_contact_epoch`` or applying a setpoint to a DER the operator
        believes is isolated.

        The returned status reports whether each drain completed. A drain that
        timed out does not abort the simulation -- the window is open either way
        -- but it tells the caller the isolation was not clean.
        """
        if not math.isfinite(duration_seconds) or duration_seconds <= 0:
            # An infinite or nonsensical duration would defeat the one guarantee
            # this method makes. Rejected before anything is suspended so a bad
            # call cannot half-isolate the client.
            raise ValueError(
                f"duration_seconds must be finite and greater than zero, got {duration_seconds!r}"
            )
        if not math.isfinite(drain_timeout) or drain_timeout < 0:
            # An unbounded drain would stall between arming the gates and
            # scheduling the timer that closes them -- isolated, with nothing
            # left to restore it.
            raise ValueError(
                f"drain_timeout must be finite and not negative, got {drain_timeout!r}"
            )

        async with self._comm_loss_lifecycle:
            # A generation per window. Everything that ends a window checks it,
            # so a timer or clear belonging to a previous one cannot reach in
            # and disarm the window that replaced it.
            self._comm_loss_generation += 1
            generation = self._comm_loss_generation
            # Cancelled, not awaited: an older timer may itself be waiting on
            # this lock, and awaiting it here would deadlock. The generation
            # check is what makes cancelling sufficient.
            self._cancel_comm_loss_simulation_expiry()

            expires_at = time.time() + duration_seconds
            # The sleep runs on the monotonic clock, like the rest of the
            # scheduler, so a wall-clock step cannot cut a window short or
            # stretch it. `expires_at` stays wall-clock because that is what an
            # operator reads.
            monotonic_deadline = time.monotonic() + duration_seconds

            notifications_drained = True
            try:
                if self._notification_server is not None:
                    notifications_drained = await self._notification_server.suspend(
                        drain_timeout=drain_timeout
                    )
                requests_drained = await self._http.arm_comm_loss_simulation(
                    expires_at=expires_at, drain_timeout=drain_timeout
                )
            except BaseException:
                # Both gates close before their drains await, so a cancellation
                # or failure part-way through would otherwise leave the client
                # isolated with no timer to restore it -- the one outcome this
                # must never produce. Roll back to a client that was never
                # isolated, and let the caller see the failure.
                self._http.comm_loss_simulation.abort()
                if self._notification_server is not None:
                    self._notification_server.resume()
                raise

            self._comm_loss_simulation_expiry = asyncio.create_task(
                self._expire_comm_loss_simulation(monotonic_deadline, generation),
                name="comm-loss-simulation-expiry",
            )

            logger.warning(
                "Comms-loss simulation started for %.0fs (operator-triggered); "
                "upstream traffic suppressed",
                duration_seconds,
            )
            status = self.comm_loss_simulation_status()
            status["requests_drained"] = requests_drained
            status["notifications_drained"] = notifications_drained
            return status

    async def clear_comm_loss_simulation(self) -> dict[str, Any]:
        """End the simulation and drive recovery, without waiting for a probe.

        Clearing the gate is not the same event as leaving comms-loss mode: the
        detector only exits once a request actually reaches the server, which
        waits on the connectivity heartbeat and then a probe tick -- some two
        and a half minutes of apparent inactivity on default settings. So
        recovery is driven here rather than left to the schedule.
        """
        async with self._comm_loss_lifecycle:
            return await self._clear_comm_loss_simulation_locked()

    async def _clear_comm_loss_simulation_locked(self) -> dict[str, Any]:
        """Restore the link and recover. Caller holds the lifecycle lock."""
        self._cancel_comm_loss_simulation_expiry()
        self._http.disarm_comm_loss_simulation()
        if self._notification_server is not None:
            self._notification_server.resume()

        simulation = self._http.comm_loss_simulation
        try:
            await self._recover_after_simulation()
        except Exception as exc:  # noqa: BLE001 - reported, not swallowed
            # Recovery re-polls schedules against a live head-end and can fail
            # for reasons that have nothing to do with the simulation. The
            # window is closed either way; the operator needs to see that
            # recovery did not complete rather than a bare "recovered".
            logger.warning("Comms-loss simulation cleared, but recovery failed: %s", exc)
            simulation.note_recovery_failed(str(exc))
            return self.comm_loss_simulation_status()

        if self._comms_loss.active:
            simulation.note_recovery_failed(
                "still in loss-of-communications mode; rediscovery did not complete"
            )
        else:
            simulation.note_recovered()
        return self.comm_loss_simulation_status()

    async def _recover_after_simulation(self) -> None:
        """Bring the client back up to date once the gate reopens.

        Full comms-loss recovery when the detector actually entered the mode.
        Otherwise a plain rediscovery, which still has to happen: notifications
        were accepted and dropped for the whole window, so a schedule or control
        change delivered only by notification is missing even though the
        detector never fired -- a short window, or a client with comms-loss
        detection disabled entirely.
        """
        if self._comms_loss.active:
            await self.recover_from_comms_loss_now()
            return
        await self._scheduler.run_exclusive("comms_loss", self._rediscover_after_simulation)

    async def _rediscover_after_simulation(self) -> None:
        if not await self.trigger_rediscovery():
            logger.warning(
                "Comms-loss simulation cleared, but rediscovery did not complete; "
                "resources delivered by notification during the window may be stale"
            )

    async def recover_from_comms_loss_now(self) -> None:
        """Run comms-loss recovery immediately, serialized with the scheduler.

        Takes the same per-key lock as the scheduled comms-loss probe. Calling
        ``_recover_from_comms_loss`` directly would bypass it, letting a
        scheduled tick overlap this one -- and recovery checks for and possibly
        re-POSTs the client's own EndDevice before it reaches the rediscovery
        lock, so two overlapping runs can register twice against a live
        head-end.

        The active check happens inside the lock, so a tick that recovered while
        this call was waiting is not undone by a second, redundant recovery.
        """

        async def _run() -> None:
            if not self._comms_loss.active:
                return
            await self._recover_from_comms_loss()

        await self._scheduler.run_exclusive("comms_loss", _run)

    def comm_loss_simulation_status(self) -> dict[str, Any]:
        """Report the simulation window and its recovery phase."""
        return self._http.comm_loss_simulation.status()

    def _cancel_comm_loss_simulation_expiry(self) -> None:
        """Stop the expiry timer, unless the caller *is* that timer.

        A timer clearing the window it owns must not cancel itself part-way
        through recovery, and the handle has to stay set so shutdown still has
        something to wait on while that recovery is in flight.
        """
        task = self._comm_loss_simulation_expiry
        if task is None or task is asyncio.current_task():
            return
        task.cancel()
        self._comm_loss_simulation_expiry = None

    async def _stop_comm_loss_simulation_expiry(self) -> None:
        """Cancel the expiry timer and wait for it to actually stop.

        Shutdown uses this rather than the bare cancel: a task merely asked to
        cancel has not run yet, and this one must be finished before the
        resources its clear path would touch are torn down.
        """
        task = self._comm_loss_simulation_expiry
        self._comm_loss_simulation_expiry = None
        if task is not None:
            task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await task

    async def _expire_comm_loss_simulation(self, deadline: float, generation: int) -> None:
        """Close the window on time, with no operator action.

        Sleeps to an absolute monotonic deadline rather than for the requested
        duration: the deadline is fixed when the caller asks, but this task only
        starts once both drains have finished, so sleeping the full duration
        would hold the window open past the ``expires`` already reported.
        """
        try:
            await asyncio.sleep(max(0.0, deadline - time.monotonic()))
        except asyncio.CancelledError:
            return
        async with self._comm_loss_lifecycle:
            if generation != self._comm_loss_generation:
                # A newer window owns the client now. This timer woke too late
                # to matter, and clearing here would disarm somebody else's
                # simulation.
                return
            if not self._http.comm_loss_simulation.active:
                return
            logger.info("Comms-loss simulation reached its deadline; restoring")
            # Cleared through the same path an operator would use, so expiry and
            # a manual clear cannot diverge. The handle is deliberately left
            # set: shutdown waits on it, and this recovery talks to the head-end.
            await self._clear_comm_loss_simulation_locked()

    async def _poll_dcap(self) -> None:
        await self._poll_with_404_recovery(self._do_poll_dcap)

    async def _poll_edev(self) -> None:
        await self._poll_with_404_recovery(self._do_poll_edev)

    async def _poll_fsa(self) -> None:
        await self._poll_with_404_recovery(self._do_poll_fsa)

    async def _poll_derp(self) -> None:
        await self._poll_with_404_recovery(self._do_poll_derp)

    async def _poll_tariff(self) -> None:
        await self._poll_with_404_recovery(self._do_poll_tariff)

    async def _poll_time(self) -> None:
        await self._poll_with_404_recovery(self._do_poll_time)

    async def _do_poll_dcap(self) -> None:
        from py20305.models.sep.sep import DeviceCapability

        logger.debug("Polling dcap")
        self._state.dcap = await self._http.get(self._dcap_path, DeviceCapability)

        # Late MirrorUsagePoint adoption: if the link was absent at discovery but
        # this DeviceCapability now advertises one (the server brought the
        # MirrorUsagePoint function set online after connect), trigger rediscovery
        # so telemetry is wired up without a restart. Rediscovery re-extracts
        # mup_list_href and re-attempts telemetry setup via the structural-change
        # path (_restart_device_telemetry).
        if (
            self._state.mup_list_href is None
            and _extract_href(self._state.dcap.mirror_usage_point_list_link)
            and self._on_structural_change
        ):
            logger.info("MirrorUsagePointListLink now advertised, triggering rediscovery")
            await self._on_structural_change()

    async def _do_poll_edev(self) -> None:
        logger.debug(
            "Polling edev: refreshing end device lists for %d devices", len(self._state.end_devices)
        )
        await refresh_end_device_lists(self._http, self._state)

    def _subscribed_fsa_hrefs(self) -> set[str]:
        """FSA list hrefs with an active subscription -- rule (r): don't poll them."""
        mgr = self._subscription_manager
        if mgr is None:
            return set()
        return {
            s.subscribed_resource for s in mgr.active_subscriptions if s.resource_type == "FSAList"
        }

    async def _do_poll_fsa(self) -> None:
        logger.debug(
            "Polling fsa: refreshing FSA lists for %d end devices", len(self._state.end_devices)
        )
        # Skip FSA lists we're subscribed to (rule r); a cancelled FSA is still
        # polled while its still-subscribed siblings are not.
        removed_programs = await refresh_function_set_assignments(
            self._http, self._state, skip_hrefs=self._subscribed_fsa_hrefs()
        )

        # Gap 3: cancel events from removed FSA programs (IEEE 8.8.3)
        if removed_programs:
            logger.info(
                "Cancelling events from %d program(s) due to FSA removal",
                len(removed_programs),
            )
            for program_href in removed_programs:
                self._event_processor.cancel_program(program_href)
                # Clean up state
                self._state.der_programs.pop(program_href, None)

    async def _do_poll_derp(self) -> None:
        logger.debug(
            "Polling derp: refreshing DER programs for %d end devices", len(self._state.end_devices)
        )
        removed_programs = await refresh_der_programs(self._http, self._state) or []
        for program_href in removed_programs:
            self._event_processor.cancel_program(program_href)
        n_progs = len(self._state.der_programs)
        logger.debug("Polling derp: refreshing controls for %d programs", n_progs)
        await refresh_der_controls(self._http, self._state)
        for href in list(self._state.der_programs):
            await self._event_processor.process_controls(href)

    async def _do_poll_tariff(self) -> None:
        """Pricing poll: re-walk the tariff tree, then relay any active-interval
        price change to connectors. No-op unless pricing is enabled."""
        await refresh_tariffs(self._http, self._state)
        await self._tariff_processor.process_tariffs()

    async def _do_poll_derp_targeted(self, subscribed_resource: str) -> bool:
        """Fetch controls for the single program that owns *subscribed_resource*.

        Returns True if a targeted refresh was performed, False if the
        resource could not be resolved to a known program (caller should
        fall back to full re-poll).
        """
        program_href = self._state.find_program_for_resource(subscribed_resource)
        if program_href is None:
            logger.warning(
                "Cannot resolve %s to a known program, falling back to full re-poll",
                subscribed_resource,
            )
            return False

        logger.debug(
            "Targeted DERP refresh for program %s (from %s)",
            program_href,
            subscribed_resource,
        )
        found = await refresh_der_controls_for_program(self._http, self._state, program_href)
        if found:
            await self._event_processor.process_controls(program_href)
        return found

    async def _do_poll_time(self) -> None:
        """Refresh every Time observation the timebase serves, global and per-FSA.

        Both scopes have to be renewed here. Event classification and timer
        firing read the FSA-scoped offset (IEEE 2030.5 §9.2.3), which
        :meth:`ServerTimebase.offset` prefers over the global one, so a per-FSA
        observation left at its discovery-time value is the offset the client
        actually runs on. Refreshing only the global scope produced a client
        whose reported drift was correct and whose scheduling was not: on a host
        with no NTP, events drifted with the local clock while ``/status`` looked
        healthy the whole time.

        Each distinct href is fetched once and observed for every scope that
        names it. A server commonly points DeviceCapability and every FSA at the
        same ``/tm``, and polling it once per FSA would multiply this poll by the
        FSA count for no new information.

        Isolation is deliberately partial. An error that ``_poll_with_404_recovery``
        exists to act on -- a redirect, or anything the global href answers other
        than 200 -- is re-raised even when other scopes refreshed fine, because
        swallowing it is the same silent failure this method fixes, one level up:
        the global Time href moves, the refresh quietly stops, and the recovery
        that would repair it never runs. The FSA hrefs keep the tolerant path
        that discovery already gives them, where an absent Time resource
        (404/204) means this FSA has none rather than that the server is broken.
        """
        from py20305.models.sep.sep import Time

        # href -> the scopes reading it; None is the global scope. The global
        # href goes in first, so when several fail it is its error that reaches
        # the recovery path -- the one that says the client's own view of the
        # server moved, rather than one FSA's endpoint being down.
        scopes: dict[str, list[str | None]] = {}
        if self._state.time_href:
            scopes.setdefault(self._state.time_href, []).append(None)
        for fsa_href, time_href in self._state.fsa_time_hrefs.items():
            scopes.setdefault(time_href, []).append(fsa_href)

        refreshed = 0
        recoverable: Exception | None = None  # must reach _poll_with_404_recovery
        last_error: Exception | None = None  # any real failure, for a total loss
        for time_href, hrefs in scopes.items():
            is_global = None in hrefs
            try:
                resource = await self._http.get(time_href, Time)
            except Exception as exc:
                if _is_absent_fsa_time(exc, is_global=is_global):
                    # Same reading discovery gives it: this FSA publishes no
                    # Time resource. Not a failure, and not a reason to
                    # rediscover on every poll for the life of the deployment.
                    logger.debug("FSA Time at %s absent; keeping prior observation", time_href)
                    continue
                # Otherwise one unreachable Time resource must not cost the
                # others their refresh: dropping the global observation because
                # an FSA's endpoint is down is how a local failure becomes a
                # fleet-wide clock problem. The stale observation stays until it
                # succeeds.
                last_error = exc
                if recoverable is None and _needs_poll_recovery(exc, is_global=is_global):
                    recoverable = exc
                logger.warning("Time refresh failed at %s; keeping prior observation", time_href)
                continue
            refreshed += 1
            for scope in hrefs:
                if scope is None:
                    self._state.time = resource
                else:
                    self._state.fsa_time[scope] = (time_href, resource)
                observe_time_resource(self._timebase, resource, fsa_href=scope, href=time_href)

        if recoverable is not None:
            # Raised even though other scopes refreshed. Selecting it up front
            # rather than re-raising whichever error came last is what keeps
            # this independent of href iteration order: a global 404 behind an
            # FSA connection error still reaches the rediscovery branch.
            raise recoverable

        if refreshed == 0 and last_error is not None:
            # Isolating each href must not turn a total failure into a success.
            # The caller treats a clean return as proof of contact and clears the
            # failure streak on it, so swallowing the last error here would keep
            # a server that answers nothing looking reachable -- and suppress the
            # rediscovery that recovers from it. Re-raise the original so the
            # 404 and 204 handling upstream still sees the status it needs.
            raise last_error

    async def _note_successful_contact(self) -> None:
        """Clear the poll-failure streak after the server is reachable again.

        Runs after a successful poll *or* a benign 204 (both prove the server
        responded). Gap 4 connectivity-recovery, two regimes:
          * Short failure (count < threshold) -- likely a network blip or brief
            5xx; re-check subscription state but keep current discovery state.
          * Sustained failure (count >= threshold) -- the server probably
            restarted; its in-memory subscription store, telemetry
            MUPs, and resource paths may all differ now, so trigger full
            rediscovery rather than leaving the client half-wired with dead
            subscriptions.
        """
        if self._poll_failure_count == 0:
            return
        recovered_after = self._poll_failure_count
        self._poll_failure_count = 0
        if recovered_after >= _SERVER_RESTART_DETECTION_THRESHOLD:
            logger.info(
                "Server reachable again after %d consecutive poll failures -- "
                "treating as upstream restart and triggering full rediscovery",
                recovered_after,
            )
            # A genuine reconnect: the server's subscription store is presumed gone,
            # so drop any honoured status=1 suppression and re-attempt every desired
            # resource from a clean session. This is the only place the
            # suppression is cleared -- not on every rediscovery.
            if self._subscription_manager is not None:
                self._subscription_manager.clear_resubscribe_suppression()
            await self.trigger_rediscovery()
        else:
            logger.info(
                "Connectivity recovered after %d failure(s), re-checking subscriptions",
                recovered_after,
            )
            if self._subscription_manager:
                await self._auto_subscribe()

    async def _poll_with_404_recovery(
        self,
        poll_fn: Callable[[], Awaitable[None]],
    ) -> None:
        try:
            await poll_fn()
            await self._note_successful_contact()

        except Sep2NoContentError:
            # 204 No Content on a polled resource is benign (the resource is
            # present but empty) -- nothing to act on. Must be caught before the
            # generic Sep2ProtocolError below (it's a subclass) so an unexpected
            # 204 can never crash the poll loop (CSIP [GEN.037]). It is still a
            # *successful* contact, though, so it clears the failure streak and
            # runs the same recovery path as a normal poll.
            logger.debug("Poll got 204 No Content; treating as empty (benign)")
            await self._note_successful_contact()
        except Sep2RedirectError as exc:
            # Persistent redirect (the retry loop already retried transient ones):
            # resources moved, so re-discover from dcap (IEEE 2030.5 §5.5.2.7)
            # rather than propagating an error.
            logger.info("Poll got redirect -> %s; triggering rediscovery", exc.location or "(none)")
            self._poll_failure_count = 0
            await self.trigger_rediscovery()
        except Sep2ProtocolError as exc:
            self._poll_failure_count += 1
            if exc.status_code == 404:
                logger.warning("Got 404 during poll, triggering rediscovery")
                self._poll_failure_count = 0
                await self.trigger_rediscovery()
            else:
                raise
        except Sep2PayloadError as exc:
            # Server replied 200 but body was malformed/empty. This is an
            # operational issue with the upstream, not a client bug. Log a
            # clean one-line warning -- the message already includes the
            # path, length, and snippet -- and continue polling.
            self._poll_failure_count += 1
            logger.warning("Poll skipped due to unparseable response: %s", exc)
        except Exception:
            self._poll_failure_count += 1
            raise

    async def poll_now(self) -> int:
        """Trigger an immediate DER program control poll.

        Unlike trigger_rediscovery(), this does NOT reset the event processor
        or re-run discovery. It simply refreshes DERControl lists for all
        known programs and processes any new/changed controls through the
        existing event processor.

        Returns the number of programs polled.
        """
        logger.info("Immediate derp poll requested")
        await self._do_poll_derp()
        n = len(self._state.der_programs)
        logger.info("Immediate derp poll complete: %d program(s)", n)
        return n

    def _update_poll_suppression(self) -> None:
        """Suppress or unsuppress polls based on active subscriptions."""
        mgr = self._subscription_manager
        if mgr is None:
            return

        active_poll_keys = {
            _RESOURCE_TYPE_TO_POLL_KEY[rt]
            for rt in mgr.subscribed_resource_types()
            if rt in _RESOURCE_TYPE_TO_POLL_KEY
        }

        # FSA granularity: the "fsa" poll refreshes every
        # end device's FSA list, but suppression is per-key -- so a single
        # subscribed FSA would suppress polling of a *different*, cancelled FSA on
        # another EndDevice, losing its structural changes. Keep the fsa poll
        # running whenever a discovered FSA list is NOT actively subscribed (it
        # needs poll-only observation); _do_poll_fsa skips the subscribed ones
        # (rule r), so the still-subscribed siblings are never redundantly polled.
        subscribed_fsa = self._subscribed_fsa_hrefs()
        all_fsa = {
            _extract_href(ed.device.function_set_assignments_list_link)
            for ed in self._state.end_devices.values()
        }
        all_fsa.discard(None)
        if all_fsa - subscribed_fsa:
            active_poll_keys.discard("fsa")

        # The tariff processor is poll-driven for active-interval transitions: it
        # has no event timers (unlike DERControl), so an interval that simply
        # becomes active by wall-clock produces no server change and no
        # notification. Keep the tariff poll running even when subscribed -- the
        # subscription only accelerates pickup of schedule *edits* between polls.
        active_poll_keys.discard("tariff")

        # Suppress keys covered by subscriptions
        for key in active_poll_keys:
            self._scheduler.suppress(key)

        # Unsuppress keys no longer covered
        for key in _RESOURCE_TYPE_TO_POLL_KEY.values():
            if key not in active_poll_keys:
                self._scheduler.unsuppress(key)

    def _find_agg_sub_href(self) -> str | None:
        """Find the client device's SubscriptionListLink href."""
        for edev_state in self._state.end_devices.values():
            if edev_state.subscription_list_href:
                return edev_state.subscription_list_href
        return None

    def _compute_desired_subscriptions(self) -> set[tuple[str, str]]:
        """Return the set of (subscribed_resource, resource_type) tuples to subscribe to.

        Pure function over DiscoveredState -- no I/O. Used by both
        ``_auto_subscribe`` and ``trigger_rediscovery`` reconciliation.
        """
        desired: set[tuple[str, str]] = set()

        # EndDeviceList
        edev_list_href = _extract_href(
            self._state.dcap.end_device_list_link if self._state.dcap else None
        )
        if edev_list_href:
            desired.add((edev_list_href, "EndDeviceList"))

        for edev_state in self._state.end_devices.values():
            # FSAList (only if server advertises subscribable)
            fsa_href = _extract_href(edev_state.device.function_set_assignments_list_link)
            if fsa_href and edev_state.fsa_list_subscribable:
                desired.add((fsa_href, "FSAList"))

            # DERProgramList per FSA (only if server advertises subscribable)
            if edev_state.derp_list_subscribable:
                for fsa in edev_state.fsa_list:
                    derp_href = _extract_href(fsa.derprogram_list_link)
                    if derp_href:
                        desired.add((derp_href, "DERProgramList"))

        # DERControlList and DefaultDERControl per program
        for derp_state in self._state.der_programs.values():
            derc_href = _extract_href(derp_state.program.dercontrol_list_link)
            if derc_href and derp_state.derc_list_subscribable:
                desired.add((derc_href, "DERControlList"))
            dderc_href = _extract_href(derp_state.program.default_dercontrol_link)
            if dderc_href and derp_state.dderc_subscribable:
                desired.add((dderc_href, "DefaultDERControl"))

        # TimeTariffIntervalList per rate component (Pricing; only if subscribable).
        # tariff_profiles is only populated when pricing is enabled, but gate
        # explicitly so intent is clear.
        if self._state.pricing_enabled:
            for tp_state in self._state.tariff_profiles.values():
                for rc_state in tp_state.rate_components:
                    if rc_state.tti_list_subscribable and rc_state.tti_list_href:
                        desired.add((rc_state.tti_list_href, "TimeTariffIntervalList"))

        return desired

    async def _auto_subscribe(self) -> None:
        """Subscribe to key resources using the client's SubscriptionListLink."""
        mgr = self._subscription_manager
        if mgr is None:
            return

        agg_sub_href = self._find_agg_sub_href()
        if not agg_sub_href:
            # The server didn't expose a SubscriptionListLink on any discovered
            # EndDevice, so there's nowhere to POST a Subscription resource and the
            # client silently falls back to polling. CSIP CORE-018 Setup #1
            # requires servers that support subscriptions to include this link on
            # the client's EndDevice; without it the ``subscribable`` attribute is
            # an empty advertisement. Warn once so this isn't invisible.
            if not self._no_sub_link_warned:
                logger.warning(
                    "Subscriptions disabled: server did not expose a "
                    "SubscriptionListLink on any discovered EndDevice. Falling "
                    "back to polling. (CSIP CORE-018 Setup #1 requires the "
                    "EndDevice for this client's SFDI/LFDI to include a "
                    "SubscriptionListLink when the server supports subscriptions.)"
                )
                self._no_sub_link_warned = True
            return

        if self._no_sub_link_warned:
            logger.info(
                "Subscriptions now available: SubscriptionListLink discovered at %s",
                agg_sub_href,
            )
            self._no_sub_link_warned = False

        for resource, rtype in self._compute_desired_subscriptions():
            if mgr.resubscribe_suppressed(resource):
                # Server cancelled this resource's subscription (status=1); honour
                # it (poll-only) instead of re-subscribing. Cleared on rediscovery.
                continue
            await mgr.subscribe(agg_sub_href, resource, rtype)

    async def _handle_notification(self, notification: object) -> None:
        """Process a received IEEE 2030.5 notification.

        Routes STATUS_DEFAULT notifications based on subscribed_resource path:
        - DERControlList changes (/derc): re-poll DERPs (fast path)
        - EndDevice/FSA/DERProgram changes: trigger full rediscovery (structural)

        STATUS_RESOURCE_DELETED with an EndDevice path triggers device removal.
        """
        from py20305.models.sep.sep import Notification

        if not isinstance(notification, Notification):
            logger.warning("Unexpected notification type: %s", type(notification))
            return

        mgr = self._subscription_manager
        if mgr is None:
            return

        # Dedup: skip if we already processed this resource recently.
        # Only STATUS_DEFAULT (data-change) notifications are deduped. Lifecycle
        # statuses (cancel/moved/definition-changed/deleted) are distinct one-shot
        # events, never duplicates of a preceding change -- deduping them swallows
        # a cancellation that arrives right after a change for the same resource
        # (conformance test ERR-002: add_fsa_err status=0 then cancel_subscriptions status=1
        # on /edev/2/fsa), so the cancellation is never honoured. Re-processing a
        # lifecycle notification is idempotent, so exempting them is safe.
        # Exception 1: a structural notification arriving while rediscovery is running must
        # not be completely lost — set the pending flag so the running rediscovery loops
        # once more and picks up the change.
        # Exception 2: DERControlList / DefaultDERControl notifications skip dedup entirely.
        # The targeted-fetch handler is a single GET against one URL, so it's cheap to repeat,
        # and burst notifications (e.g. conformance test BASIC-015's 23-call
        # cancel_derc) must each be
        # processed — deduping them means we only see snapshots from the earliest notifications
        # and miss the trailing state where the burst's later cancellations apply.
        # Classify by the subscription's topology-derived resource_type (naming-
        # independent); the path heuristics inside _is_control/_is_structural are
        # only the fallback when this subscription isn't tracked.
        resource_type = mgr.resource_type_for(notification.subscription_uri)
        if (
            notification.status == STATUS_DEFAULT
            and not self._is_control(notification.subscribed_resource, resource_type)
            and mgr.is_duplicate_notification(notification.subscribed_resource)
        ):
            if self._is_structural(notification.subscribed_resource, resource_type) and (
                self._rediscovery_lock.locked()
            ):
                self._rediscovery_pending = True
            logger.debug("Ignoring duplicate notification for %s", notification.subscribed_resource)
            return

        # List/non-list filter: non-list subscriptions should not fire on list changes
        if not mgr.should_process_notification(
            notification.subscription_uri, notification.subscribed_resource
        ):
            logger.debug(
                "Ignoring notification (list/non-list filter) for %s",
                notification.subscribed_resource,
            )
            return

        import time

        if notification.status == STATUS_DEFAULT:
            mgr.record_notification(
                StoredNotification(
                    subscribed_resource=notification.subscribed_resource,
                    status=notification.status,
                    subscription_uri=notification.subscription_uri,
                    created_at=time.time(),
                )
            )
            await self._route_default_notification(notification.subscribed_resource, resource_type)
        elif notification.status == STATUS_CANCELLED:
            # Server cancelled with no further info (rule n, status=1). Honour it:
            # stop treating the sub as active and DON'T auto-re-subscribe -- fall
            # back to polling the resource (permitted once the sub is inactive per
            # rule r). The suppression persists across rediscovery and is cleared
            # only on a genuine reconnect, so an unrelated rediscovery
            # can't undo the honoured cancellation.
            mgr.mark_cancelled(notification.subscription_uri)
            mgr.remove_notifications_for(notification.subscribed_resource)
            mgr.suppress_resubscribe(notification.subscribed_resource)
            self._update_poll_suppression()
        elif notification.status == STATUS_RESOURCE_MOVED:
            # Resource URI changed -> the server terminated the sub (rule h,
            # status=2). Record the new location for visibility, then rediscover
            # so we re-subscribe at the resource's new URI (discovery walks the
            # tree; new_resource_uri alone lacks the resource_type/context).
            mgr.mark_cancelled(notification.subscription_uri)
            self._update_poll_suppression()
            if notification.new_resource_uri:
                mgr.record_notification(
                    StoredNotification(
                        subscribed_resource=notification.subscribed_resource,
                        status=notification.status,
                        subscription_uri=notification.subscription_uri,
                        created_at=time.time(),
                        new_resource_uri=notification.new_resource_uri,
                    )
                )
            if self._on_structural_change:
                await self._on_structural_change()
            else:
                await self.trigger_rediscovery()
        elif notification.status == STATUS_DEFINITION_CHANGED:
            mgr.mark_cancelled(notification.subscription_uri)
            self._update_poll_suppression()
        elif notification.status == STATUS_RESOURCE_DELETED:
            mgr.mark_cancelled(notification.subscription_uri)
            self._update_poll_suppression()
            await self._handle_resource_deleted(notification.subscribed_resource)

    def _is_structural(self, resource_path: str, resource_type: str | None = None) -> bool:
        """Classify a notification as structural (EndDevice/FSA -> rediscovery).

        Prefers the subscription's topology-derived ``resource_type`` (naming-
        independent); falls back to the path heuristic only when the type is
        unknown (the notification's subscription isn't tracked locally).
        """
        if resource_type is not None:
            return resource_type in _STRUCTURAL_RESOURCE_TYPES
        return self._is_structural_resource(resource_path)

    def _is_control(self, resource_path: str, resource_type: str | None = None) -> bool:
        """Classify a notification as a control change (DERControl -> targeted fetch).

        Prefers ``resource_type``; falls back to the path heuristic when unknown.
        """
        if resource_type is not None:
            return resource_type in _CONTROL_RESOURCE_TYPES
        return self._is_control_resource(resource_path)

    def _is_structural_resource(self, resource_path: str) -> bool:
        """Return True if the resource path represents a structural change.

        Path-heuristic fallback used when the notification's subscription type
        isn't known. Only EndDevice and FSA changes require full rediscovery (they
        alter the device mapping); DERProgram and DERControl changes are handled
        by a lighter re-poll via ``_do_poll_derp``.
        """
        path = resource_path.rstrip("/")
        parts = path.strip("/").split("/")
        if not parts:
            return True

        has_control = "derc" in parts or "dderc" in parts
        has_program = "derp" in parts

        # EndDeviceList (/edev) or a single EndDevice (/edev/<id>). Match ``edev``
        # as a path segment near the end rather than at position 0, so a server
        # base-path prefix (e.g. /api/v2/edev) still resolves. Deeper paths
        # (.../edev/<id>/fsa, .../der/...) fall through to the fsa branch.
        if "edev" in parts:
            edev_idx = parts.index("edev")
            if len(parts) - edev_idx <= 2:
                return True

        # FSA changes (contains /fsa but not /derp or /derc)
        return "fsa" in parts and not has_program and not has_control

    @staticmethod
    def _is_control_resource(resource_path: str) -> bool:
        """Return True if the path is a DERControlList or DefaultDERControl.

        These are candidates for targeted fetch (single-program refresh)
        rather than a full re-poll of all programs.
        """
        path = resource_path.rstrip("/")
        return path.endswith("/derc") or path.endswith("/dderc")

    def _is_tariff(self, resource_path: str, resource_type: str | None = None) -> bool:
        """Classify a notification as a Pricing change (tariff -> tariff re-poll).

        Prefers ``resource_type``; falls back to the path heuristic when unknown.
        """
        if resource_type is not None:
            return resource_type in _TARIFF_RESOURCE_TYPES
        return self._is_tariff_resource(resource_path)

    @staticmethod
    def _is_tariff_resource(resource_path: str) -> bool:
        """Return True if the path is a TariffProfileList or TimeTariffIntervalList.

        Path-heuristic fallback used when the subscription's type isn't tracked.
        The tariff tree convention is ``/tp/<n>/rc/<n>/tti`` (see the Pricing plan).
        """
        path = resource_path.rstrip("/")
        return path.endswith("/tti") or path.endswith("/tp")

    async def _route_default_notification(
        self, subscribed_resource: str, resource_type: str | None = None
    ) -> None:
        """Route a STATUS_DEFAULT notification based on resource type.

        ``resource_type`` (the subscription's topology-derived type) drives the
        classification when known; otherwise the path heuristic is the fallback.
        """
        if self._is_structural(subscribed_resource, resource_type):
            logger.info(
                "Structural change notification for %s, triggering rediscovery",
                subscribed_resource,
            )
            if self._on_structural_change:
                await self._on_structural_change()
            else:
                await self.trigger_rediscovery()
        else:
            # Skip re-poll if rediscovery is in progress — the device mapping
            # is in flux and process_controls would produce incorrect results.
            # Rediscovery will pick up the changes when it re-fetches controls.
            if self._rediscovery_lock.locked():
                logger.debug(
                    "Skipping control re-poll for %s (rediscovery in progress)",
                    subscribed_resource,
                )
                return

            # Pricing change: refresh the tariff tree and relay any active-interval
            # price change (parallel to the DER path; not a control or structural).
            if self._is_tariff(subscribed_resource, resource_type):
                logger.info(
                    "Tariff change notification for %s, re-polling tariffs",
                    subscribed_resource,
                )
                await self._do_poll_tariff()
                return

            # Targeted fetch for DERControlList / DefaultDERControl notifications
            if self._is_control(subscribed_resource, resource_type):
                logger.info(
                    "Control change notification for %s, targeted fetch",
                    subscribed_resource,
                )
                # Record ancestor paths in the dedup cache BEFORE the targeted
                # fetch's awaits. The server sends a sibling DERProgramList
                # notification alongside the DERControlList one (its
                # DERControlListLink.all count changed); recording ancestry up
                # front suppresses that sibling instead of letting it race ahead
                # into a redundant full re-poll during our await window. Recorded
                # unconditionally: on the rare targeted-fetch failure we fall
                # through to a full re-poll below, which covers the sibling too.
                mgr = self._subscription_manager
                if mgr:
                    mgr.record_notification_ancestry(subscribed_resource)
                if await self._do_poll_derp_targeted(subscribed_resource):
                    return
                logger.info(
                    "Targeted fetch failed for %s, falling back to full re-poll",
                    subscribed_resource,
                )

            logger.info(
                "Control change notification for %s, re-polling DERPs",
                subscribed_resource,
            )
            await self._do_poll_derp()

    async def _handle_resource_deleted(self, subscribed_resource: str) -> None:
        """Handle STATUS_RESOURCE_DELETED notification.

        If the deleted resource is an EndDevice, call the device-removal callback.
        Otherwise trigger rediscovery.
        """
        path = subscribed_resource.rstrip("/")
        parts = path.strip("/").split("/")

        # EndDevice deletion: /edev/X (exactly 2 segments)
        if len(parts) == 2 and parts[0] == "edev":
            logger.info("EndDevice deleted: %s", subscribed_resource)
            if self._on_device_removed:
                await self._on_device_removed(subscribed_resource)
            return

        # Other resource deletions trigger rediscovery
        logger.info("Resource deleted: %s, triggering rediscovery", subscribed_resource)
        if self._on_structural_change:
            await self._on_structural_change()
        else:
            await self.trigger_rediscovery()

    async def trigger_rediscovery(self) -> bool:
        """Re-run discovery and restart polling. Safe to call concurrently.

        If called while rediscovery is already running, sets a pending flag so
        the running rediscovery will loop once more after completing. This ensures
        structural change notifications received mid-run are not silently dropped.

        Returns True when a rediscovery pass completed (state rebuilt, polls
        re-armed); False when discovery raised (polls are still re-armed so the
        heartbeat can recover) or when the call coalesced into an already-running
        rediscovery. Most callers ignore the result; the comms-loss recovery path
        uses it to avoid clearing the outage gate on a failed re-poll.
        """
        if self._rediscovery_lock.locked():
            self._rediscovery_pending = True
            logger.debug("Rediscovery already in progress, will re-run after completion")
            return False

        async with self._rediscovery_lock:
            passes = 0
            # A status=1 resubscribe-suppression is deliberately NOT cleared here.
            # Rediscovery is frequent (structural-change notifications trigger it),
            # so clearing on every pass would undo an honoured server cancellation
            # -- a status=1 that lands just before an unrelated rediscovery would be
            # wiped and the resource re-subscribed. The suppression is
            # cleared only on a genuine reconnect (upstream-restart detection in
            # _note_successful_contact), where the server's subscription store is
            # presumed gone. The cancelled resource is still observed via polling
            # (mark_cancelled unsuppresses its poll key), so honouring the
            # cancellation does not lose changes.
            while True:
                self._rediscovery_pending = False
                passes += 1
                logger.info("Rediscovery: resetting polls (preserving event processor)")
                await self._scheduler.cancel_all()
                self._scheduler = PollScheduler(heartbeat_enabled=self._heartbeat_enabled)
                # Cancel renewal task (reconciliation handles subscriptions)
                if self._renewal_task and not self._renewal_task.done():
                    self._renewal_task.cancel()
                    with contextlib.suppress(asyncio.CancelledError):
                        await self._renewal_task
                    self._renewal_task = None
                # Remember active programs before clearing state
                previous_programs = set(self._state.der_programs.keys())
                # Signal that state is being rebuilt — event timers that fire
                # during this window will wait until discover() repopulates it.
                self._state_ready.clear()
                # A rediscovery is commonly triggered by a *reconnect* (upstream
                # restart detection). If the server flaps -- recovers just long
                # enough to trigger this, then drops again -- a GET here (discover
                # or a refresh) raises. The scheduler was already rebuilt empty
                # above, so we MUST still re-arm it in that case: otherwise no
                # poll and no connectivity heartbeat fire, and server_alive is
                # stuck false forever while telemetry keeps last_contact fresh but
                # never chain-validates. _start_polls() therefore runs on both the
                # success and failure paths.
                try:
                    try:
                        # Clear discovered state but preserve subscriptions for reconciliation.
                        # Registration PIN verification is a one-shot startup check
                        # (done in connect()), so it's intentionally not repeated on
                        # rediscovery -- no registration_pins passed here.
                        self._state.clear()
                        await discover(
                            self._http,
                            self._state,
                            dcap_path=self._dcap_path,
                        )
                    finally:
                        # Always unblock timer callbacks — if discover() failed they will
                        # see empty state (harmless no-ops) rather than hanging forever.
                        self._state_ready.set()
                    if self._subscription_manager:
                        desired = self._compute_desired_subscriptions()
                        agg_sub_href = self._find_agg_sub_href()
                        if agg_sub_href and desired:
                            # renew_kept=False: rediscovery must not re-POST kept
                            # subscriptions. Re-POSTing makes some servers echo a
                            # current-state notification, which we classify as a
                            # structural change and rediscover again -- an infinite
                            # loop. The periodic renewal task owns subscription
                            # lifetime; rediscovery only reconciles membership.
                            result = await self._subscription_manager.reconcile(
                                agg_sub_href, desired, renew_kept=False
                            )
                            logger.info(
                                "Subscription reconciliation: kept=%d cancelled=%d"
                                " created=%d renewed=%d",
                                result.kept,
                                result.cancelled,
                                result.created,
                                result.renewed,
                            )
                            # renew_kept=False keeps subs without verifying them, so a
                            # subscription whose server-side URI has drifted would still
                            # be reported under its stale local URI. Adopt the server's
                            # current URIs so post-rediscovery state matches the server
                            # (a stale URI otherwise 404s on a later cancel/renew).
                            # reestablish_missing=False: adopt-only, no re-POST -- a
                            # re-POST here risks the rediscovery<->echo loop.
                            await self._subscription_manager.reconcile_with_server(
                                agg_sub_href, reestablish_missing=False
                            )
                        else:
                            # No subscription list or no desired subs; cancel remaining
                            await self._subscription_manager.cancel_all()
                        self._update_poll_suppression()
                        if self._subscription_manager.active_subscriptions:
                            self._renewal_task = asyncio.create_task(
                                self._subscription_manager.start_renewal_task(
                                    self._shutdown_event,
                                    interval_seconds=self._renewal_interval_seconds,
                                )
                            )
                    # Re-fetch FSA and program hierarchy to catch resources added between
                    # discover() and subscription creation (same race as DERControl refresh).
                    await refresh_function_set_assignments(self._http, self._state)
                    pruned_programs = await refresh_der_programs(self._http, self._state) or []
                    if self._subscription_manager:
                        await self._auto_subscribe()
                        self._update_poll_suppression()
                    # Cancel events from programs that no longer exist after rediscovery
                    current_programs = set(self._state.der_programs.keys())
                    all_removed = (previous_programs - current_programs) | set(pruned_programs)
                    for removed in all_removed:
                        self._event_processor.cancel_program(removed)
                    # Re-fetch DER controls before processing: data from discover() may be
                    # stale if controls were written during subscription reconciliation.
                    await refresh_der_controls(self._http, self._state)
                    # Process controls discovered during rediscovery immediately
                    for href in list(self._state.der_programs):
                        await self._event_processor.process_controls(href)
                    self._start_polls()
                    logger.info("Rediscovery complete")
                except Exception as exc:
                    # Re-arm the scheduler so the connectivity heartbeat (and polls)
                    # keep running: the heartbeat's validating GET recovers
                    # server_alive, and a later successful poll re-triggers
                    # rediscovery to rebuild state. Return rather than re-raise --
                    # the caller is often a poll loop or the reconnect path, and the
                    # empty-scheduler condition is now healed.
                    from py20305.diagnostics import report

                    report(
                        "warnings",
                        f"Rediscovery failed ({exc}); re-armed polls so the "
                        "connectivity heartbeat can recover server_alive",
                        source="client",
                        dedup_key="rediscovery:failed",
                        details={"error": str(exc)},
                        exc_info=True,
                    )
                    self._start_polls()
                    return False
                if not self._rediscovery_pending:
                    break
                if passes >= _MAX_REDISCOVERY_PASSES:
                    self._rediscovery_pending = False
                    from py20305.diagnostics import report

                    report(
                        "warnings",
                        f"Rediscovery still pending after {passes} passes; stopping to "
                        "avoid an unbounded rediscovery/notification loop. A server that "
                        "echoes a structural notification on every subscription write can "
                        "cause this.",
                        source="client",
                        dedup_key="rediscovery:max_passes",
                        details={"passes": passes},
                    )
                    logger.warning(
                        "Rediscovery still pending after %d passes; stopping to avoid an "
                        "unbounded rediscovery<->notification loop",
                        passes,
                    )
                    break
                logger.info(
                    "Re-running rediscovery (structural change received during previous run)"
                )
        if self._on_rediscovered:
            # Outside the pass loop: a re-run means the state is still moving,
            # and a listener that rescheduled work per pass would do it twice.
            await self._on_rediscovered()
        return True

    async def shutdown(self, timeout: float = 10.0) -> None:
        """Gracefully stop polling, event timers, and close the HTTP session."""
        self._shutdown_event.set()
        # Before anything is torn down. An expiry firing later would call
        # clear_comm_loss_simulation() against a stopped notification server and
        # scheduler, and its recovery would reopen the HTTP session this method
        # is about to close.
        await self._stop_comm_loss_simulation_expiry()
        if self._renewal_task and not self._renewal_task.done():
            self._renewal_task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await self._renewal_task
        if self._subscription_manager:
            await self._subscription_manager.cancel_all()
        if self._notification_server:
            await self._notification_server.stop()
        await self._event_processor.shutdown()
        await self._tariff_processor.shutdown()
        await self._scheduler.cancel_all(timeout=timeout)
        await self._http.close()

    async def __aenter__(self) -> CsipClient:
        return self

    async def __aexit__(
        self,
        _exc_type: type[BaseException] | None,
        _exc_val: BaseException | None,
        _exc_tb: TracebackType | None,
    ) -> None:
        await self.shutdown()

dispatcher property

dispatcher: ControlDispatcher

The dispatcher this client applies controls through.

audit property

audit: AuditEmitter

The audit trail emitter. Attach a forwarder and configure it to enable.

subscription_manager property

subscription_manager: SubscriptionManager | None

The subscription manager, when subscriptions are wired.

attach_subscriptions

attach_subscriptions(manager: SubscriptionManager, notification_server: NotificationServer | None = None) -> None

Wire subscribe/notify into this client after construction.

The manager needs this client's transport, which exists only once the client does, so a caller building both cannot pass them to __init__ -- construction order forces attachment to come second. Attach before connect(): auto-subscription and poll suppression consult the manager during connection.

Source code in src/py20305/client/csip_client.py

def attach_subscriptions(
    self,
    manager: SubscriptionManager,
    notification_server: NotificationServer | None = None,
) -> None:
    """Wire subscribe/notify into this client after construction.

    The manager needs this client's transport, which exists only once the
    client does, so a caller building both cannot pass them to
    ``__init__`` -- construction order forces attachment to come second.
    Attach before ``connect()``: auto-subscription and poll suppression
    consult the manager during connection.
    """
    self._subscription_manager = manager
    if notification_server is not None:
        self._notification_server = notification_server
        notification_server.on_notification = self._handle_notification

set_on_structural_change

set_on_structural_change(callback: Callable[[], Awaitable[None]] | None) -> None

Set the structural-change callback after construction.

The same construction-order problem as attach_subscriptions, one step later: telemetry is started once discovery has produced the hrefs it posts to, so what wants to hear about a later structural change does not exist when this client is built.

Note what this callback replaces. Where a structural notification would otherwise call trigger_rediscovery(), it calls this instead, so a callback that does not rediscover leaves the state it was called about unrefreshed. To act on rebuilt state, use set_on_rediscovered.

Source code in src/py20305/client/csip_client.py

def set_on_structural_change(
    self, callback: Callable[[], Awaitable[None]] | None
) -> None:
    """Set the structural-change callback after construction.

    The same construction-order problem as ``attach_subscriptions``, one
    step later: telemetry is started once discovery has produced the hrefs
    it posts to, so what wants to hear about a later structural change does
    not exist when this client is built.

    Note what this callback replaces. Where a structural notification would
    otherwise call ``trigger_rediscovery()``, it calls this instead, so a
    callback that does not rediscover leaves the state it was called about
    unrefreshed. To act on rebuilt state, use ``set_on_rediscovered``.
    """
    self._on_structural_change = callback

set_on_rediscovered

set_on_rediscovered(callback: Callable[[], Awaitable[None]] | None) -> None

Set a callback to run after any rediscovery pass that completed.

Every path that rebuilds discovered state ends here -- a structural notification, 404 recovery, comms-loss recovery, a subscription lost, an operator's request through the management API -- so anything holding hrefs from the previous state refreshes once, in one place, rather than each trigger remembering to say so.

Not called when discovery raised, because the state was not rebuilt, nor when the call coalesced into a running rediscovery, because that one calls it on completion.

Source code in src/py20305/client/csip_client.py

def set_on_rediscovered(self, callback: Callable[[], Awaitable[None]] | None) -> None:
    """Set a callback to run after any rediscovery pass that completed.

    Every path that rebuilds discovered state ends here -- a structural
    notification, 404 recovery, comms-loss recovery, a subscription lost,
    an operator's request through the management API -- so anything holding
    hrefs from the previous state refreshes once, in one place, rather than
    each trigger remembering to say so.

    Not called when discovery raised, because the state was not rebuilt,
    nor when the call coalesced into a running rediscovery, because that
    one calls it on completion.
    """
    self._on_rediscovered = callback

run_redirect_probe async

run_redirect_probe() -> dict[str, Any]

Run ERR-001 HTTP-to-HTTPS redirect probe using this client's session.

Source code in src/py20305/client/csip_client.py

async def run_redirect_probe(self) -> dict[str, Any]:
    """Run ERR-001 HTTP-to-HTTPS redirect probe using this client's session."""
    from py20305.client.redirect_probe import run_redirect_probe

    return await run_redirect_probe(self._http.host, self._http)

register_end_device async

register_end_device(*, lfdi: str, device_category: int | None = None, edev_list_href: str | None = None, check_duplicate: bool = True) -> str | None

In-band registration: POST a managed EndDevice to the server's EndDeviceList and return the Location (the new /edev/{id} href).

lfdi is a 40-hex string; sFDI is derived from it (never free-entered). device_category is an optional DeviceCategoryType bitmap. The target list defaults to the discovered dcap.end_device_list_link unless edev_list_href overrides it.

Raises ValueError for a missing EndDeviceList href or, when check_duplicate, an LFDI the server already lists (IEEE 2030.5 §8.5.3: clients SHALL NOT POST a duplicate EndDevice). Raises Sep2ProtocolError on a server rejection status.

Source code in src/py20305/client/csip_client.py

async def register_end_device(
    self,
    *,
    lfdi: str,
    device_category: int | None = None,
    edev_list_href: str | None = None,
    check_duplicate: bool = True,
) -> str | None:
    """In-band registration: POST a managed EndDevice to the server's
    EndDeviceList and return the ``Location`` (the new ``/edev/{id}`` href).

    ``lfdi`` is a 40-hex string; ``sFDI`` is derived from it (never
    free-entered). ``device_category`` is an optional DeviceCategoryType
    bitmap. The target list defaults to the discovered
    ``dcap.end_device_list_link`` unless ``edev_list_href`` overrides it.

    Raises ``ValueError`` for a missing EndDeviceList href or, when
    ``check_duplicate``, an LFDI the server already lists (IEEE 2030.5 §8.5.3:
    clients SHALL NOT POST a duplicate EndDevice). Raises ``Sep2ProtocolError``
    on a server rejection status.
    """
    from py20305.models.sep.sep import (
        DeviceCategoryType,
        EndDevice,
        Sfditype,
        TimeType,
    )
    from py20305.security.identity import compute_sfdi

    lfdi_bytes = bytes.fromhex(lfdi)
    href = edev_list_href or _extract_href(
        self._state.dcap.end_device_list_link if self._state.dcap else None
    )
    if not href:
        raise ValueError(
            "No EndDeviceList href -- the server's DeviceCapability has not been discovered"
        )

    if check_duplicate and await self._server_lists_end_device(lfdi, edev_list_href=href):
        raise ValueError(f"An EndDevice with LFDI {lfdi} is already registered")

    end_device = EndDevice(
        l_fdi=lfdi_bytes,
        s_fdi=Sfditype(value=compute_sfdi(lfdi)),
        changed_time=TimeType(value=int(self._timebase.now())),
    )
    if device_category is not None:
        if not 0 <= device_category < 2**32:
            raise ValueError("device_category must be between 0 and 4294967295 (32-bit bitmap)")
        end_device.device_category = DeviceCategoryType(
            value=device_category.to_bytes(4, "big")
        )
    return await self._http.post(href, end_device)

connect async

connect() -> None

Run initial resource discovery and auto-subscribe if configured.

Source code in src/py20305/client/csip_client.py

async def connect(self) -> None:
    """Run initial resource discovery and auto-subscribe if configured."""
    if self._notification_server and not self._notification_server.running:
        await self._notification_server.start()
    await discover(
        self._http,
        self._state,
        registration_pins=self._registration_pins,
        dcap_path=self._dcap_path,
    )
    # Capture the client's own LFDI for comms-loss self-reregistration.
    # Cert-derived and stable across reset_session, so a single capture holds.
    if self._own_lfdi is None:
        self._own_lfdi = self._http.client_lfdi
    if self._subscription_manager:
        if self._subscription_manager.active_subscriptions:
            valid, removed = await self._subscription_manager.validate_restored_subscriptions()
            logger.info("Subscription validation: %d valid, %d removed", valid, removed)
        await self._auto_subscribe()
        self._update_poll_suppression()
        if self._subscription_manager.active_subscriptions:
            self._renewal_task = asyncio.create_task(
                self._subscription_manager.start_renewal_task(
                    self._shutdown_event, interval_seconds=self._renewal_interval_seconds
                )
            )

run async

run() -> None

Start polling and block until shutdown is signaled.

Source code in src/py20305/client/csip_client.py

async def run(self) -> None:
    """Start polling and block until shutdown is signaled."""
    logger.info("Starting poll loop")
    # Process controls from initial discovery immediately
    for href in list(self._state.der_programs):
        await self._event_processor.process_controls(href)
    self._start_polls()
    await self._shutdown_event.wait()
    logger.info("Poll loop exiting (shutdown signaled)")

simulate_comm_loss async

simulate_comm_loss(duration_seconds: float, *, drain_timeout: float = 5.0) -> dict[str, Any]

Cut this client off from the head-end for duration_seconds.

Mechanism, not policy: whether a simulation is permitted at all, and how long a window may reasonably run, are the caller's decisions. This method does what it is told and guarantees only that the window closes.

Notifications are suspended before the transport gate closes, so a control cannot slip in through the inbound path while the outbound one is being shut. Both then drain: a request or handler already running would otherwise finish after the window opened, refreshing last_contact_epoch or applying a setpoint to a DER the operator believes is isolated.

The returned status reports whether each drain completed. A drain that timed out does not abort the simulation -- the window is open either way -- but it tells the caller the isolation was not clean.

Source code in src/py20305/client/csip_client.py

async def simulate_comm_loss(
    self, duration_seconds: float, *, drain_timeout: float = 5.0
) -> dict[str, Any]:
    """Cut this client off from the head-end for ``duration_seconds``.

    Mechanism, not policy: whether a simulation is permitted at all, and how
    long a window may reasonably run, are the caller's decisions. This
    method does what it is told and guarantees only that the window closes.

    Notifications are suspended before the transport gate closes, so a
    control cannot slip in through the inbound path while the outbound one
    is being shut. Both then drain: a request or handler already running
    would otherwise finish after the window opened, refreshing
    ``last_contact_epoch`` or applying a setpoint to a DER the operator
    believes is isolated.

    The returned status reports whether each drain completed. A drain that
    timed out does not abort the simulation -- the window is open either way
    -- but it tells the caller the isolation was not clean.
    """
    if not math.isfinite(duration_seconds) or duration_seconds <= 0:
        # An infinite or nonsensical duration would defeat the one guarantee
        # this method makes. Rejected before anything is suspended so a bad
        # call cannot half-isolate the client.
        raise ValueError(
            f"duration_seconds must be finite and greater than zero, got {duration_seconds!r}"
        )
    if not math.isfinite(drain_timeout) or drain_timeout < 0:
        # An unbounded drain would stall between arming the gates and
        # scheduling the timer that closes them -- isolated, with nothing
        # left to restore it.
        raise ValueError(
            f"drain_timeout must be finite and not negative, got {drain_timeout!r}"
        )

    async with self._comm_loss_lifecycle:
        # A generation per window. Everything that ends a window checks it,
        # so a timer or clear belonging to a previous one cannot reach in
        # and disarm the window that replaced it.
        self._comm_loss_generation += 1
        generation = self._comm_loss_generation
        # Cancelled, not awaited: an older timer may itself be waiting on
        # this lock, and awaiting it here would deadlock. The generation
        # check is what makes cancelling sufficient.
        self._cancel_comm_loss_simulation_expiry()

        expires_at = time.time() + duration_seconds
        # The sleep runs on the monotonic clock, like the rest of the
        # scheduler, so a wall-clock step cannot cut a window short or
        # stretch it. `expires_at` stays wall-clock because that is what an
        # operator reads.
        monotonic_deadline = time.monotonic() + duration_seconds

        notifications_drained = True
        try:
            if self._notification_server is not None:
                notifications_drained = await self._notification_server.suspend(
                    drain_timeout=drain_timeout
                )
            requests_drained = await self._http.arm_comm_loss_simulation(
                expires_at=expires_at, drain_timeout=drain_timeout
            )
        except BaseException:
            # Both gates close before their drains await, so a cancellation
            # or failure part-way through would otherwise leave the client
            # isolated with no timer to restore it -- the one outcome this
            # must never produce. Roll back to a client that was never
            # isolated, and let the caller see the failure.
            self._http.comm_loss_simulation.abort()
            if self._notification_server is not None:
                self._notification_server.resume()
            raise

        self._comm_loss_simulation_expiry = asyncio.create_task(
            self._expire_comm_loss_simulation(monotonic_deadline, generation),
            name="comm-loss-simulation-expiry",
        )

        logger.warning(
            "Comms-loss simulation started for %.0fs (operator-triggered); "
            "upstream traffic suppressed",
            duration_seconds,
        )
        status = self.comm_loss_simulation_status()
        status["requests_drained"] = requests_drained
        status["notifications_drained"] = notifications_drained
        return status

clear_comm_loss_simulation async

clear_comm_loss_simulation() -> dict[str, Any]

End the simulation and drive recovery, without waiting for a probe.

Clearing the gate is not the same event as leaving comms-loss mode: the detector only exits once a request actually reaches the server, which waits on the connectivity heartbeat and then a probe tick -- some two and a half minutes of apparent inactivity on default settings. So recovery is driven here rather than left to the schedule.

Source code in src/py20305/client/csip_client.py

async def clear_comm_loss_simulation(self) -> dict[str, Any]:
    """End the simulation and drive recovery, without waiting for a probe.

    Clearing the gate is not the same event as leaving comms-loss mode: the
    detector only exits once a request actually reaches the server, which
    waits on the connectivity heartbeat and then a probe tick -- some two
    and a half minutes of apparent inactivity on default settings. So
    recovery is driven here rather than left to the schedule.
    """
    async with self._comm_loss_lifecycle:
        return await self._clear_comm_loss_simulation_locked()

recover_from_comms_loss_now async

recover_from_comms_loss_now() -> None

Run comms-loss recovery immediately, serialized with the scheduler.

Takes the same per-key lock as the scheduled comms-loss probe. Calling _recover_from_comms_loss directly would bypass it, letting a scheduled tick overlap this one -- and recovery checks for and possibly re-POSTs the client's own EndDevice before it reaches the rediscovery lock, so two overlapping runs can register twice against a live head-end.

The active check happens inside the lock, so a tick that recovered while this call was waiting is not undone by a second, redundant recovery.

Source code in src/py20305/client/csip_client.py

async def recover_from_comms_loss_now(self) -> None:
    """Run comms-loss recovery immediately, serialized with the scheduler.

    Takes the same per-key lock as the scheduled comms-loss probe. Calling
    ``_recover_from_comms_loss`` directly would bypass it, letting a
    scheduled tick overlap this one -- and recovery checks for and possibly
    re-POSTs the client's own EndDevice before it reaches the rediscovery
    lock, so two overlapping runs can register twice against a live
    head-end.

    The active check happens inside the lock, so a tick that recovered while
    this call was waiting is not undone by a second, redundant recovery.
    """

    async def _run() -> None:
        if not self._comms_loss.active:
            return
        await self._recover_from_comms_loss()

    await self._scheduler.run_exclusive("comms_loss", _run)

comm_loss_simulation_status

comm_loss_simulation_status() -> dict[str, Any]

Report the simulation window and its recovery phase.

Source code in src/py20305/client/csip_client.py

def comm_loss_simulation_status(self) -> dict[str, Any]:
    """Report the simulation window and its recovery phase."""
    return self._http.comm_loss_simulation.status()

poll_now async

poll_now() -> int

Trigger an immediate DER program control poll.

Unlike trigger_rediscovery(), this does NOT reset the event processor or re-run discovery. It simply refreshes DERControl lists for all known programs and processes any new/changed controls through the existing event processor.

Returns the number of programs polled.

Source code in src/py20305/client/csip_client.py

async def poll_now(self) -> int:
    """Trigger an immediate DER program control poll.

    Unlike trigger_rediscovery(), this does NOT reset the event processor
    or re-run discovery. It simply refreshes DERControl lists for all
    known programs and processes any new/changed controls through the
    existing event processor.

    Returns the number of programs polled.
    """
    logger.info("Immediate derp poll requested")
    await self._do_poll_derp()
    n = len(self._state.der_programs)
    logger.info("Immediate derp poll complete: %d program(s)", n)
    return n

trigger_rediscovery async

trigger_rediscovery() -> bool

Re-run discovery and restart polling. Safe to call concurrently.

If called while rediscovery is already running, sets a pending flag so the running rediscovery will loop once more after completing. This ensures structural change notifications received mid-run are not silently dropped.

Returns True when a rediscovery pass completed (state rebuilt, polls re-armed); False when discovery raised (polls are still re-armed so the heartbeat can recover) or when the call coalesced into an already-running rediscovery. Most callers ignore the result; the comms-loss recovery path uses it to avoid clearing the outage gate on a failed re-poll.

Source code in src/py20305/client/csip_client.py

async def trigger_rediscovery(self) -> bool:
    """Re-run discovery and restart polling. Safe to call concurrently.

    If called while rediscovery is already running, sets a pending flag so
    the running rediscovery will loop once more after completing. This ensures
    structural change notifications received mid-run are not silently dropped.

    Returns True when a rediscovery pass completed (state rebuilt, polls
    re-armed); False when discovery raised (polls are still re-armed so the
    heartbeat can recover) or when the call coalesced into an already-running
    rediscovery. Most callers ignore the result; the comms-loss recovery path
    uses it to avoid clearing the outage gate on a failed re-poll.
    """
    if self._rediscovery_lock.locked():
        self._rediscovery_pending = True
        logger.debug("Rediscovery already in progress, will re-run after completion")
        return False

    async with self._rediscovery_lock:
        passes = 0
        # A status=1 resubscribe-suppression is deliberately NOT cleared here.
        # Rediscovery is frequent (structural-change notifications trigger it),
        # so clearing on every pass would undo an honoured server cancellation
        # -- a status=1 that lands just before an unrelated rediscovery would be
        # wiped and the resource re-subscribed. The suppression is
        # cleared only on a genuine reconnect (upstream-restart detection in
        # _note_successful_contact), where the server's subscription store is
        # presumed gone. The cancelled resource is still observed via polling
        # (mark_cancelled unsuppresses its poll key), so honouring the
        # cancellation does not lose changes.
        while True:
            self._rediscovery_pending = False
            passes += 1
            logger.info("Rediscovery: resetting polls (preserving event processor)")
            await self._scheduler.cancel_all()
            self._scheduler = PollScheduler(heartbeat_enabled=self._heartbeat_enabled)
            # Cancel renewal task (reconciliation handles subscriptions)
            if self._renewal_task and not self._renewal_task.done():
                self._renewal_task.cancel()
                with contextlib.suppress(asyncio.CancelledError):
                    await self._renewal_task
                self._renewal_task = None
            # Remember active programs before clearing state
            previous_programs = set(self._state.der_programs.keys())
            # Signal that state is being rebuilt — event timers that fire
            # during this window will wait until discover() repopulates it.
            self._state_ready.clear()
            # A rediscovery is commonly triggered by a *reconnect* (upstream
            # restart detection). If the server flaps -- recovers just long
            # enough to trigger this, then drops again -- a GET here (discover
            # or a refresh) raises. The scheduler was already rebuilt empty
            # above, so we MUST still re-arm it in that case: otherwise no
            # poll and no connectivity heartbeat fire, and server_alive is
            # stuck false forever while telemetry keeps last_contact fresh but
            # never chain-validates. _start_polls() therefore runs on both the
            # success and failure paths.
            try:
                try:
                    # Clear discovered state but preserve subscriptions for reconciliation.
                    # Registration PIN verification is a one-shot startup check
                    # (done in connect()), so it's intentionally not repeated on
                    # rediscovery -- no registration_pins passed here.
                    self._state.clear()
                    await discover(
                        self._http,
                        self._state,
                        dcap_path=self._dcap_path,
                    )
                finally:
                    # Always unblock timer callbacks — if discover() failed they will
                    # see empty state (harmless no-ops) rather than hanging forever.
                    self._state_ready.set()
                if self._subscription_manager:
                    desired = self._compute_desired_subscriptions()
                    agg_sub_href = self._find_agg_sub_href()
                    if agg_sub_href and desired:
                        # renew_kept=False: rediscovery must not re-POST kept
                        # subscriptions. Re-POSTing makes some servers echo a
                        # current-state notification, which we classify as a
                        # structural change and rediscover again -- an infinite
                        # loop. The periodic renewal task owns subscription
                        # lifetime; rediscovery only reconciles membership.
                        result = await self._subscription_manager.reconcile(
                            agg_sub_href, desired, renew_kept=False
                        )
                        logger.info(
                            "Subscription reconciliation: kept=%d cancelled=%d"
                            " created=%d renewed=%d",
                            result.kept,
                            result.cancelled,
                            result.created,
                            result.renewed,
                        )
                        # renew_kept=False keeps subs without verifying them, so a
                        # subscription whose server-side URI has drifted would still
                        # be reported under its stale local URI. Adopt the server's
                        # current URIs so post-rediscovery state matches the server
                        # (a stale URI otherwise 404s on a later cancel/renew).
                        # reestablish_missing=False: adopt-only, no re-POST -- a
                        # re-POST here risks the rediscovery<->echo loop.
                        await self._subscription_manager.reconcile_with_server(
                            agg_sub_href, reestablish_missing=False
                        )
                    else:
                        # No subscription list or no desired subs; cancel remaining
                        await self._subscription_manager.cancel_all()
                    self._update_poll_suppression()
                    if self._subscription_manager.active_subscriptions:
                        self._renewal_task = asyncio.create_task(
                            self._subscription_manager.start_renewal_task(
                                self._shutdown_event,
                                interval_seconds=self._renewal_interval_seconds,
                            )
                        )
                # Re-fetch FSA and program hierarchy to catch resources added between
                # discover() and subscription creation (same race as DERControl refresh).
                await refresh_function_set_assignments(self._http, self._state)
                pruned_programs = await refresh_der_programs(self._http, self._state) or []
                if self._subscription_manager:
                    await self._auto_subscribe()
                    self._update_poll_suppression()
                # Cancel events from programs that no longer exist after rediscovery
                current_programs = set(self._state.der_programs.keys())
                all_removed = (previous_programs - current_programs) | set(pruned_programs)
                for removed in all_removed:
                    self._event_processor.cancel_program(removed)
                # Re-fetch DER controls before processing: data from discover() may be
                # stale if controls were written during subscription reconciliation.
                await refresh_der_controls(self._http, self._state)
                # Process controls discovered during rediscovery immediately
                for href in list(self._state.der_programs):
                    await self._event_processor.process_controls(href)
                self._start_polls()
                logger.info("Rediscovery complete")
            except Exception as exc:
                # Re-arm the scheduler so the connectivity heartbeat (and polls)
                # keep running: the heartbeat's validating GET recovers
                # server_alive, and a later successful poll re-triggers
                # rediscovery to rebuild state. Return rather than re-raise --
                # the caller is often a poll loop or the reconnect path, and the
                # empty-scheduler condition is now healed.
                from py20305.diagnostics import report

                report(
                    "warnings",
                    f"Rediscovery failed ({exc}); re-armed polls so the "
                    "connectivity heartbeat can recover server_alive",
                    source="client",
                    dedup_key="rediscovery:failed",
                    details={"error": str(exc)},
                    exc_info=True,
                )
                self._start_polls()
                return False
            if not self._rediscovery_pending:
                break
            if passes >= _MAX_REDISCOVERY_PASSES:
                self._rediscovery_pending = False
                from py20305.diagnostics import report

                report(
                    "warnings",
                    f"Rediscovery still pending after {passes} passes; stopping to "
                    "avoid an unbounded rediscovery/notification loop. A server that "
                    "echoes a structural notification on every subscription write can "
                    "cause this.",
                    source="client",
                    dedup_key="rediscovery:max_passes",
                    details={"passes": passes},
                )
                logger.warning(
                    "Rediscovery still pending after %d passes; stopping to avoid an "
                    "unbounded rediscovery<->notification loop",
                    passes,
                )
                break
            logger.info(
                "Re-running rediscovery (structural change received during previous run)"
            )
    if self._on_rediscovered:
        # Outside the pass loop: a re-run means the state is still moving,
        # and a listener that rescheduled work per pass would do it twice.
        await self._on_rediscovered()
    return True

shutdown async

shutdown(timeout: float = 10.0) -> None

Gracefully stop polling, event timers, and close the HTTP session.

Source code in src/py20305/client/csip_client.py

async def shutdown(self, timeout: float = 10.0) -> None:
    """Gracefully stop polling, event timers, and close the HTTP session."""
    self._shutdown_event.set()
    # Before anything is torn down. An expiry firing later would call
    # clear_comm_loss_simulation() against a stopped notification server and
    # scheduler, and its recovery would reopen the HTTP session this method
    # is about to close.
    await self._stop_comm_loss_simulation_expiry()
    if self._renewal_task and not self._renewal_task.done():
        self._renewal_task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await self._renewal_task
    if self._subscription_manager:
        await self._subscription_manager.cancel_all()
    if self._notification_server:
        await self._notification_server.stop()
    await self._event_processor.shutdown()
    await self._tariff_processor.shutdown()
    await self._scheduler.cancel_all(timeout=timeout)
    await self._http.close()

py20305.client.tls

TLS configuration for IEEE 2030.5 connections.

TlsConfig dataclass

TLS client configuration for IEEE 2030.5.

Source code in src/py20305/client/tls.py

@dataclass(frozen=True)
class TlsConfig:
    """TLS client configuration for IEEE 2030.5."""

    client_cert: Path
    client_key: Path
    ca_cert: Path
    check_hostname: bool = True
    ciphers: str = _IEEE_2030_5_CIPHERS
    additional_ciphers: tuple[str, ...] = ()
    # Send the client's cert-derived LFDI as an HTTP header on every outbound
    # request. Off by default: IEEE 2030.5 §6.11.7.2 says servers SHOULD
    # derive the LFDI from the client cert, so the header is non-spec and
    # can trip strict-validation WAFs. Enable for utility deployments
    # fronted by a TLS-terminating proxy that strips the client cert before
    # the backend sees it.
    send_lfdi_header: bool = False
    # Header name to carry the LFDI under. IEEE 2030.5 doesn't standardize
    # this -- known peer conventions vary, so the name is configurable. The
    # default ("LFDI") matches the only peer convention we have evidence for;
    # override per peer (e.g. "X-LFDI") when a peer expects a different header.
    # Only honored when send_lfdi_header is True.
    lfdi_header_name: str = "LFDI"

CertChainError

Bases: SSLCertVerificationError

IEEE 2030.5 certificate chain validation failure.

Raised when the peer's certificate chain violates IEEE 2030.5 Section 6.11.3 requirements that OpenSSL does not enforce natively.

Source code in src/py20305/client/tls.py

class CertChainError(ssl.SSLCertVerificationError):
    """IEEE 2030.5 certificate chain validation failure.

    Raised when the peer's certificate chain violates IEEE 2030.5
    Section 6.11.3 requirements that OpenSSL does not enforce natively.
    """

build_cipher_string

build_cipher_string(config: TlsConfig) -> str

Combine the baseline cipher list with any operator-supplied additions.

Each addition must name exactly one suite, which is checked against OpenSSL rather than assumed. An OpenSSL cipher string is an expression language, so a single entry can rewrite the policy instead of adding to it: ALL widens it to everything, !aNULL removes what the baseline exists to require, and plain-looking aliases such as RSA or AES quietly expand to dozens of suites. The result of this function goes straight to SSLContext.set_ciphers, which accepts all of that silently, so nothing downstream would catch it.

Raises:

Type	Description
ValueError	If an addition does not name exactly one TLS 1.2 suite.

Source code in src/py20305/client/tls.py

def build_cipher_string(config: TlsConfig) -> str:
    """Combine the baseline cipher list with any operator-supplied additions.

    Each addition must name exactly one suite, which is checked against
    OpenSSL rather than assumed. An OpenSSL cipher string is an expression
    language, so a single entry can rewrite the policy instead of adding to
    it: ``ALL`` widens it to everything, ``!aNULL`` removes what the baseline
    exists to require, and plain-looking aliases such as ``RSA`` or ``AES``
    quietly expand to dozens of suites. The result of this function goes
    straight to ``SSLContext.set_ciphers``, which accepts all of that
    silently, so nothing downstream would catch it.

    Raises:
        ValueError: If an addition does not name exactly one TLS 1.2 suite.
    """
    if not config.additional_ciphers:
        return config.ciphers

    for name in config.additional_ciphers:
        if name not in _known_tls12_suites():
            raise ValueError(
                f"additional_ciphers entries must each name exactly one TLS 1.2 "
                f"cipher suite; {name!r} is unknown to OpenSSL or selects a group, "
                f"which would widen the IEEE 2030.5 baseline rather than add to it"
            )

    extras = ":".join(config.additional_ciphers)
    return f"{config.ciphers}:{extras}"

create_ssl_context

create_ssl_context(config: TlsConfig) -> ssl.SSLContext

Create an SSL context from a TlsConfig.

Source code in src/py20305/client/tls.py

def create_ssl_context(config: TlsConfig) -> ssl.SSLContext:
    """Create an SSL context from a TlsConfig."""
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ctx.minimum_version = ssl.TLSVersion.TLSv1_2
    ctx.maximum_version = ssl.TLSVersion.TLSv1_2
    ctx.verify_mode = ssl.CERT_REQUIRED
    ctx.check_hostname = config.check_hostname
    if config.additional_ciphers:
        logger.warning(
            "TLS cipher policy relaxed beyond IEEE 2030.5 baseline; additional ciphers: %s",
            ", ".join(config.additional_ciphers),
        )
    ctx.set_ciphers(build_cipher_string(config))
    ctx.load_cert_chain(str(config.client_cert), str(config.client_key))
    ctx.load_verify_locations(str(config.ca_cert))
    ctx.verify_flags |= ssl.VERIFY_X509_PARTIAL_CHAIN
    return ctx

verify_ieee2030_5_chain

verify_ieee2030_5_chain(chain_der: list[bytes]) -> None

Validate a verified cert chain against IEEE 2030.5 PKI requirements.

Checks intermediate CA certificates (all except leaf and root) for extensions that IEEE 2030.5 and RFC 5280 prohibit or constrain:

• Extended Key Usage MUST NOT be marked critical on CA certs (IEEE 2030.5 Section 6.11.3)
• Name Constraints MUST be critical if present (RFC 5280 Section 4.2.1.10)
• Policy Constraints MUST NOT be present (IEEE 2030.5 Section 6.11.3)

Parameters:

Name	Type	Description	Default
chain_der	list[bytes]	List of DER-encoded certificates from get_verified_chain(), ordered leaf-first (index 0 = leaf, last = root).	required

Raises:

Type	Description
CertChainError	If any CA cert violates IEEE 2030.5 requirements.

Source code in src/py20305/client/tls.py

def verify_ieee2030_5_chain(chain_der: list[bytes]) -> None:
    """Validate a verified cert chain against IEEE 2030.5 PKI requirements.

    Checks intermediate CA certificates (all except leaf and root) for
    extensions that IEEE 2030.5 and RFC 5280 prohibit or constrain:

    - Extended Key Usage MUST NOT be marked critical on CA certs
      (IEEE 2030.5 Section 6.11.3)
    - Name Constraints MUST be critical if present (RFC 5280 Section 4.2.1.10)
    - Policy Constraints MUST NOT be present (IEEE 2030.5 Section 6.11.3)

    Args:
        chain_der: List of DER-encoded certificates from get_verified_chain(),
            ordered leaf-first (index 0 = leaf, last = root).

    Raises:
        CertChainError: If any CA cert violates IEEE 2030.5 requirements.
    """
    if len(chain_der) < 2:
        return

    # Check CA certs (skip leaf at [0], skip root at [-1])
    ca_certs = chain_der[1:-1] if len(chain_der) > 2 else []

    for der_bytes in ca_certs:
        cert = x509.load_der_x509_certificate(der_bytes)
        cn = _cert_cn(cert)

        for ext in cert.extensions:
            # EKU on CA certs must not be critical
            if ext.oid == ExtensionOID.EXTENDED_KEY_USAGE and ext.critical:
                raise CertChainError(f"CA cert '{cn}' has critical ExtendedKeyUsage")

            # NameConstraints must be critical if present (RFC 5280)
            if ext.oid == ExtensionOID.NAME_CONSTRAINTS and not ext.critical:
                raise CertChainError(f"CA cert '{cn}' has non-critical NameConstraints")

            # PolicyConstraints forbidden in IEEE 2030.5 manufacturing PKI
            if ext.oid == ExtensionOID.POLICY_CONSTRAINTS:
                raise CertChainError(f"CA cert '{cn}' has PolicyConstraints (forbidden)")

py20305.client.discovery

Hierarchical IEEE 2030.5 resource discovery.

discover async

discover(client: Sep2Client, state: DiscoveredState, *, registration_pins: dict[str, int] | None = None, dcap_path: str = '/dcap') -> None

Run full hierarchical discovery: dcap -> sdev -> edev -> der -> fsa -> derp.

Populates state with all discovered resources and poll rates. Clears any existing state before starting.

Parameters:

Name	Type	Description	Default
client	Sep2Client	The Sep2Client for HTTP operations.	required
state	DiscoveredState	DiscoveredState to populate.	required
registration_pins	dict[str, int] | None	Optional mapping from LFDI hex -> expected PIN for registration verification.	None
dcap_path	str	Path to DeviceCapability resource (default: /dcap).	'/dcap'

Source code in src/py20305/client/discovery.py

async def discover(
    client: Sep2Client,
    state: DiscoveredState,
    *,
    registration_pins: dict[str, int] | None = None,
    dcap_path: str = "/dcap",
) -> None:
    """Run full hierarchical discovery: dcap -> sdev -> edev -> der -> fsa -> derp.

    Populates ``state`` with all discovered resources and poll rates.
    Clears any existing state before starting.

    Args:
        client: The Sep2Client for HTTP operations.
        state: DiscoveredState to populate.
        registration_pins: Optional mapping from LFDI hex -> expected PIN
            for registration verification.
        dcap_path: Path to DeviceCapability resource (default: /dcap).
    """
    state.clear()

    # 1. Device Capability (with raw body for CSIP-AUS detection)
    dcap, raw_dcap = await client.get_with_body(dcap_path, DeviceCapability)
    state.dcap = dcap
    state.poll_rates["dcap"] = normalize_poll_rate(dcap.poll_rate, resource_key="dcap")

    # Auto-detect CSIP-AUS mode from namespace declaration in dcap response
    if b"csipaus" in raw_dcap:
        state.csip_aus_mode = True
        logger.info("CSIP-AUS mode detected from server dcap response")

    # 2. Time
    time_href = _extract_href(dcap.time_link)
    if time_href:
        state.time_href = time_href
        state.time = await client.get(time_href, Time)
        observe_time_resource(client.timebase, state.time, href=time_href)

    # 3. SelfDevice (optional): the server's own device description. Used by
    #    clients that need the server's electrical context -- e.g. a site
    #    limit published on SelfDevice:DER:DERSettings that connected devices
    #    operate within.
    state.self_device = await _discover_self_device(client, dcap)
    if state.self_device is not None:
        state.poll_rates["sdev"] = normalize_poll_rate(
            state.self_device.device.poll_rate, resource_key="sdev"
        )

    # 4. MUP list href, for telemetry
    mup_href = _extract_href(dcap.mirror_usage_point_list_link)
    if mup_href:
        state.mup_list_href = mup_href
        logger.info("Discovered MirrorUsagePointListLink: %s", mup_href)
    else:
        logger.warning("DeviceCapability has no MirrorUsagePointListLink")

    # 5. EndDevice list
    edev_list_href = _extract_href(dcap.end_device_list_link)
    if not edev_list_href:
        logger.warning("No EndDeviceListLink in DeviceCapability")
        _set_time_poll_rate(state)
        return

    edev_pages = await client.get_list(edev_list_href, EndDeviceList)
    device_mapping = DeviceMapping()

    for page in edev_pages:
        # Capture edev poll rate from first page
        if "edev" not in state.poll_rates and page.poll_rate is not None:
            state.poll_rates["edev"] = normalize_poll_rate(page.poll_rate, resource_key="edev")

        for edev in page.end_device:
            edev_href = edev.href
            if not edev_href:
                continue

            lfdi = edev.l_fdi or b""
            edev_state = EndDeviceState(device=edev, href=edev_href, lfdi=lfdi)

            # Log event list link (for alarm posting)
            log_event_href = _extract_href(edev.log_event_list_link)
            if log_event_href:
                edev_state.log_event_list_href = log_event_href

            # Subscription list link (for subscription management)
            sub_list_href = _extract_href(edev.subscription_list_link)
            if sub_list_href:
                edev_state.subscription_list_href = sub_list_href
                # Capture the SubscriptionList's server-advertised pollRate (once,
                # for the client's own SubscriptionList) so the reconcile
                # poll runs at the rate the server prefers. normalize_poll_rate
                # falls back to the IEEE 900s default when the server omits it.
                if "sub" not in state.poll_rates:
                    try:
                        sub_list = await client.get(sub_list_href, SubscriptionList)
                        state.poll_rates["sub"] = normalize_poll_rate(
                            sub_list.poll_rate, resource_key="sub"
                        )
                    except Exception as exc:
                        # Best-effort: the pollRate is an optimization. Any failure
                        # (protocol, malformed body, transport) just leaves
                        # poll_rates["sub"] unset so the reconcile poll falls back
                        # to the IEEE 2030.5 default cadence.
                        logger.debug(
                            "Could not read SubscriptionList pollRate, using default: %s", exc
                        )

            # Registration link (for PIN verification)
            reg_href = _extract_href(edev.registration_link)
            if reg_href:
                edev_state.registration_href = reg_href

            # 5a. DER list (optional; server may advertise link without resource).
            # Contained per device: one EndDevice's unreadable DER list must not
            # abort the walk for every other device, and at startup a raise here
            # never completes connect() at all (entry_point._connect_with_retry
            # retries every Sep2Error), which costs the control path as well.
            # The cost of continuing is that this device's DER resource PUTs stay
            # unscheduled (der_resource_manager skips None hrefs), so a real
            # fault logs at error rather than warning.
            der_href = _extract_href(edev.derlist_link)
            if der_href:
                try:
                    der_list = await client.get(der_href, Derlist)
                    edev_state.ders = list(der_list.der)
                    hrefs = _der_child_hrefs(
                        edev_state.ders, der_href, legacy_multi_der=client.server_2018_compat
                    )
                    edev_state.der_status_href = hrefs.status
                    edev_state.der_capability_href = hrefs.capability
                    edev_state.der_settings_href = hrefs.settings
                    edev_state.der_availability_href = hrefs.availability
                    edev_state.current_der_controls_href = hrefs.current_controls
                except Sep2ProtocolError as exc:
                    if exc.status_code in (404, 204):
                        logger.warning(
                            "DER list at %s %s, skipping", der_href, _absent_reason(exc.status_code)
                        )
                    else:
                        logger.error(
                            "DER list at %s failed to read, no DER resource PUTs for this "
                            "device: %s",
                            der_href,
                            exc,
                        )
                except Sep2PayloadError as exc:
                    logger.error(
                        "DER list at %s did not parse, no DER resource PUTs for this device: %s",
                        der_href,
                        exc,
                    )

            # 5b. FSA list (optional; server may advertise link without resource)
            fsa_href = _extract_href(edev.function_set_assignments_list_link)
            if fsa_href:
                try:
                    fsa_pages = await client.get_list(fsa_href, FunctionSetAssignmentsList)
                except Sep2ProtocolError as exc:
                    if exc.status_code == 404:
                        logger.warning("FSA list at %s not found, skipping", fsa_href)
                        fsa_pages = []
                    else:
                        raise
                for fsa_page in fsa_pages:
                    # Capture FSA poll rate from first page
                    if "fsa" not in state.poll_rates and fsa_page.poll_rate is not None:
                        state.poll_rates["fsa"] = normalize_poll_rate(
                            fsa_page.poll_rate, resource_key="fsa"
                        )
                    if hasattr(fsa_page, "subscribable") and fsa_page.subscribable:
                        edev_state.fsa_list_subscribable = True

                    for fsa in fsa_page.function_set_assignments:
                        edev_state.fsa_list.append(fsa)
                        logger.debug("Found FSA %s for device %s", fsa.href, edev_href)

                        # Pricing function set (opt-in): tariff tree per FSA. Done
                        # before the DERProgram branch so an FSA that advertises a
                        # tariff link but no DERProgram list is still discovered.
                        if state.pricing_enabled:
                            await _discover_tariffs_for_fsa(client, fsa, state)

                        # 6. DERProgram list per FSA
                        derp_list_href = _extract_href(fsa.derprogram_list_link)
                        if not derp_list_href:
                            continue

                        try:
                            derp_pages = await client.get_list(derp_list_href, DerprogramList)
                        except Sep2ProtocolError as exc:
                            if exc.status_code == 404:
                                logger.warning(
                                    "DERP list at %s not found, skipping", derp_list_href
                                )
                                continue
                            raise
                        for derp_page in derp_pages:
                            if hasattr(derp_page, "subscribable") and derp_page.subscribable:
                                edev_state.derp_list_subscribable = True
                        await _discover_programs_from_pages(
                            client, state, device_mapping, derp_pages, edev_href, fsa.href
                        )

                        # Gap 9: fetch per-FSA Time resource if available
                        fsa_time_href = _extract_href(fsa.time_link)
                        if fsa_time_href and fsa.href:
                            # Recorded from the advertisement, before the fetch
                            # that may fail. The Time poll drives off this map,
                            # so an endpoint that is down right now is retried on
                            # the next poll instead of being forgotten until the
                            # next rediscovery.
                            state.fsa_time_hrefs[fsa.href] = fsa_time_href
                        if fsa_time_href and fsa.href and fsa.href not in state.fsa_time:
                            try:
                                fsa_time = await client.get(fsa_time_href, Time)
                                state.fsa_time[fsa.href] = (fsa_time_href, fsa_time)
                                # IEEE 9.2.3: events from this FSA's programs
                                # follow this Time resource.
                                observe_time_resource(
                                    client.timebase,
                                    fsa_time,
                                    fsa_href=fsa.href,
                                    href=fsa_time_href,
                                )
                            except Sep2ProtocolError as exc:
                                if exc.status_code in (404, 204):
                                    logger.debug(
                                        "FSA Time at %s %s",
                                        fsa_time_href,
                                        _absent_reason(exc.status_code),
                                    )
                                else:
                                    raise

            # Gap 1: dcap fallback -- when no FSA or no programs from FSAs,
            # try discovering DERPrograms directly from DeviceCapability.
            if not edev_state.fsa_list:
                dcap_derp_href = _extract_href(dcap.derprogram_list_link)
                if dcap_derp_href:
                    logger.info("No FSA for %s, falling back to dcap DERProgramListLink", edev_href)
                    try:
                        dcap_derp_pages = await client.get_list(dcap_derp_href, DerprogramList)
                        await _discover_programs_from_pages(
                            client, state, device_mapping, dcap_derp_pages, edev_href, None
                        )
                    except Sep2ProtocolError as exc:
                        if exc.status_code == 404:
                            logger.debug("dcap DERProgramList at %s not found", dcap_derp_href)
                        else:
                            raise

            # Gap 3: snapshot current FSA hrefs for removal detection on next poll
            state.previous_fsa_hrefs[edev_href] = {
                fsa.href for fsa in edev_state.fsa_list if fsa.href
            }

            # Verify registration PIN if configured. Gate the fetch on map
            # membership so EndDevices without a configured PIN don't incur an
            # extra GET on every (re)discovery in mixed-config deployments.
            if registration_pins and edev_state.registration_href:
                lfdi_hex = edev_state.lfdi.hex() if edev_state.lfdi else ""
                if lfdi_hex in registration_pins:
                    await _verify_registration(
                        client, edev_state.registration_href, lfdi_hex, registration_pins
                    )

            state.end_devices[edev_href] = edev_state

    state.device_mapping = device_mapping

    # Set poll rate for DER program control refresh if not already set from the list.
    # Per IEEE 2030.5, the DeviceCapability pollRate is the default for the
    # entire hierarchy unless a more specific rate is provided on a sub-resource.
    if state.der_programs and "derp" not in state.poll_rates:
        dcap_rate = state.poll_rates.get("dcap")
        state.poll_rates["derp"] = normalize_poll_rate(
            None, resource_key="derp", default=dcap_rate or DEFAULT_POLL_RATE
        )

    _set_time_poll_rate(state)

    logger.info(
        "Discovery complete: %d devices, %d programs",
        len(state.end_devices),
        len(state.der_programs),
    )

refresh_der_controls async

refresh_der_controls(client: Sep2Client, state: DiscoveredState) -> None

Re-fetch DER control lists, DefaultDERControl, and DERCurves for all known programs.

Unlike discover(), this does not clear state or re-run the full hierarchy walk. It refreshes the DERControl list, DefaultDERControl, and DERCurve list for each already-discovered DERProgram, so new controls, DDERC changes, and freshly-created curves are picked up without a disruptive state clear.

Source code in src/py20305/client/discovery.py

async def refresh_der_controls(client: Sep2Client, state: DiscoveredState) -> None:
    """Re-fetch DER control lists, DefaultDERControl, and DERCurves for all
    known programs.

    Unlike ``discover()``, this does not clear state or re-run the full
    hierarchy walk. It refreshes the DERControl list, DefaultDERControl, and
    DERCurve list for each already-discovered DERProgram, so new controls,
    DDERC changes, and freshly-created curves are picked up without a
    disruptive state clear.
    """
    for _, derp_state in list(state.der_programs.items()):
        # Refresh DefaultDERControl
        dderc_href = _extract_href(derp_state.program.default_dercontrol_link)
        if dderc_href:
            try:
                dderc = await client.get(dderc_href, DefaultDercontrol)
                derp_state.default_dercontrol = dderc
            except Sep2ProtocolError as exc:
                if exc.status_code in (404, 204):
                    logger.debug(
                        "DDERC at %s %s on refresh", dderc_href, _absent_reason(exc.status_code)
                    )
                    derp_state.default_dercontrol = None
                else:
                    raise
            except Sep2PayloadError as exc:
                logger.warning("Skipping DDERC refresh at %s: %s", dderc_href, exc)
            except Exception:
                logger.warning(
                    "Failed to parse DDERC at %s during refresh, skipping",
                    dderc_href,
                    exc_info=True,
                )
        else:
            # Server removed the DefaultDERControl link — clear cached value
            derp_state.default_dercontrol = None

        # Refresh DERCurve list before controls, so any new curves a fresh
        # control might reference are already cached when the translation
        # layer resolves the curve link.
        await _refresh_curves_for_program(client, derp_state)

        # Refresh DERControl list
        derc_list_href = _extract_href(derp_state.program.dercontrol_list_link)
        if not derc_list_href:
            continue

        try:
            derc_pages = await client.get_list(derc_list_href, DercontrolList)
        except Sep2ProtocolError as exc:
            if exc.status_code == 404:
                logger.warning("DERC list at %s not found during refresh", derc_list_href)
                continue
            raise
        except Sep2PayloadError as exc:
            logger.warning("Skipping DERControl refresh: %s", exc)
            continue
        except Exception:
            logger.warning(
                "Failed to parse DERControl list at %s during refresh, skipping",
                derc_list_href,
                exc_info=True,
            )
            continue

        derp_state.der_controls.clear()
        for derc_page in derc_pages:
            derp_state.der_controls.extend(derc_page.dercontrol)
        derp_state.der_controls_complete = True

    logger.debug("Refreshed DER controls for %d programs", len(state.der_programs))

refresh_der_controls_for_program async

refresh_der_controls_for_program(client: Sep2Client, state: DiscoveredState, program_href: str) -> bool

Re-fetch DERControl list and DefaultDERControl for a single program.

Returns True if the program was found and refreshed, False if program_href is not in state (e.g. removed between notification and fetch).

Source code in src/py20305/client/discovery.py

async def refresh_der_controls_for_program(
    client: Sep2Client, state: DiscoveredState, program_href: str
) -> bool:
    """Re-fetch DERControl list and DefaultDERControl for a single program.

    Returns True if the program was found and refreshed, False if
    *program_href* is not in state (e.g. removed between notification
    and fetch).
    """
    derp_state = state.der_programs.get(program_href)
    if derp_state is None:
        logger.warning("Program %s not found in state during targeted refresh", program_href)
        return False

    # Refresh DefaultDERControl
    dderc_href = _extract_href(derp_state.program.default_dercontrol_link)
    if dderc_href:
        try:
            dderc = await client.get(dderc_href, DefaultDercontrol)
            derp_state.default_dercontrol = dderc
        except Sep2ProtocolError as exc:
            if exc.status_code in (404, 204):
                logger.debug(
                    "DDERC at %s %s on targeted refresh",
                    dderc_href,
                    _absent_reason(exc.status_code),
                )
                derp_state.default_dercontrol = None
            else:
                raise
        except Sep2PayloadError as exc:
            logger.warning("Skipping DDERC targeted refresh at %s: %s", dderc_href, exc)
        except Exception:
            logger.warning(
                "Failed to parse DDERC at %s during targeted refresh, skipping",
                dderc_href,
                exc_info=True,
            )
    else:
        derp_state.default_dercontrol = None

    # Refresh DERCurve list before controls so any newly-referenced curves
    # are cached when the translation layer resolves curve links.
    await _refresh_curves_for_program(client, derp_state)

    # Refresh DERControl list
    derc_list_href = _extract_href(derp_state.program.dercontrol_list_link)
    if derc_list_href:
        try:
            derc_pages = await client.get_list(derc_list_href, DercontrolList)
        except Sep2ProtocolError as exc:
            if exc.status_code == 404:
                logger.warning("DERC list at %s not found during targeted refresh", derc_list_href)
                return True
            raise
        except Sep2PayloadError as exc:
            logger.warning("Skipping DERControl targeted refresh: %s", exc)
            return True
        except Exception:
            logger.warning(
                "Failed to parse DERControl list at %s during targeted refresh, skipping",
                derc_list_href,
                exc_info=True,
            )
            return True

        derp_state.der_controls.clear()
        for derc_page in derc_pages:
            derp_state.der_controls.extend(derc_page.dercontrol)
        derp_state.der_controls_complete = True

    logger.debug("Targeted refresh complete for program %s", program_href)
    return True

refresh_end_device_lists async

refresh_end_device_lists(client: Sep2Client, state: DiscoveredState) -> None

Re-fetch the EndDeviceList from the DeviceCapability link.

This updates EndDevice data (like changedTime) but does NOT re-discover FSA or DERProgram hierarchies. Useful for catching device-level changes without full rediscovery.

Source code in src/py20305/client/discovery.py

async def refresh_end_device_lists(client: Sep2Client, state: DiscoveredState) -> None:
    """Re-fetch the EndDeviceList from the DeviceCapability link.

    This updates EndDevice data (like changedTime) but does NOT re-discover
    FSA or DERProgram hierarchies. Useful for catching device-level changes
    without full rediscovery.
    """
    # Get the EndDeviceList href from dcap
    if not state.dcap:
        logger.warning("No DeviceCapability in state, skipping EndDevice list refresh")
        return

    edev_list_href = _extract_href(state.dcap.end_device_list_link)
    if not edev_list_href:
        logger.warning("No EndDeviceListLink in DeviceCapability")
        return

    try:
        edev_pages = await client.get_list(edev_list_href, EndDeviceList)
    except Sep2ProtocolError as exc:
        logger.warning("Failed to refresh EndDeviceList at %s: %s", edev_list_href, exc)
        return

    # Update existing devices with refreshed data
    for page in edev_pages:
        for edev in page.end_device:
            edev_href = edev.href
            if not edev_href:
                continue

            if edev_href in state.end_devices:
                # Update the device object and LFDI
                state.end_devices[edev_href].device = edev
                state.end_devices[edev_href].lfdi = edev.l_fdi or b""

    logger.debug("Refreshed EndDevice list with %d devices", len(state.end_devices))

refresh_function_set_assignments async

refresh_function_set_assignments(client: Sep2Client, state: DiscoveredState, *, skip_hrefs: set[str] | None = None) -> list[str]

Re-fetch FSA lists for all known end devices.

Returns a list of program hrefs from removed FSAs (Gap 3: IEEE 8.8.3). Callers should cancel active events from these programs.

skip_hrefs names FSA list hrefs that have an active subscription: the poll path passes these so it doesn't re-fetch a resource we're already subscribed to (IEEE 2030.5 §8.9.3.4 rule (r)). A single cancelled FSA can then be polled while its siblings stay subscription-only. Discovery/rediscovery pass nothing (a full re-sync refreshes every FSA).

Source code in src/py20305/client/discovery.py

async def refresh_function_set_assignments(
    client: Sep2Client, state: DiscoveredState, *, skip_hrefs: set[str] | None = None
) -> list[str]:
    """Re-fetch FSA lists for all known end devices.

    Returns a list of program hrefs from removed FSAs (Gap 3: IEEE 8.8.3).
    Callers should cancel active events from these programs.

    ``skip_hrefs`` names FSA list hrefs that have an active subscription: the poll
    path passes these so it doesn't re-fetch a resource we're already subscribed to
    (IEEE 2030.5 §8.9.3.4 rule (r)). A single cancelled FSA can then be polled while
    its siblings stay subscription-only. Discovery/rediscovery pass nothing (a full
    re-sync refreshes every FSA).
    """
    removed_program_hrefs: list[str] = []

    if not state.end_devices:
        return removed_program_hrefs

    for edev_href, edev_state in list(state.end_devices.items()):
        # Refresh FSA list
        fsa_href = _extract_href(edev_state.device.function_set_assignments_list_link)
        if not fsa_href:
            continue
        if skip_hrefs and fsa_href in skip_hrefs:
            continue  # active subscription -> rule (r): don't poll it

        try:
            fsa_pages = await client.get_list(fsa_href, FunctionSetAssignmentsList)
        except Sep2ProtocolError as exc:
            if exc.status_code == 404:
                logger.warning("FSA list at %s not found during refresh", fsa_href)
                continue
            raise

        edev_state.fsa_list.clear()
        for fsa_page in fsa_pages:
            edev_state.fsa_list.extend(fsa_page.function_set_assignments)

        # Gap 3: detect removed FSAs by comparing to previous snapshot
        current_fsa_hrefs = {fsa.href for fsa in edev_state.fsa_list if fsa.href}
        previous = state.previous_fsa_hrefs.get(edev_href, set())
        removed_fsas = previous - current_fsa_hrefs

        if removed_fsas:
            logger.info(
                "Detected %d removed FSA(s) for %s: %s",
                len(removed_fsas),
                edev_href,
                removed_fsas,
            )
            for removed_fsa_href in removed_fsas:
                removed_program_hrefs.extend(state.programs_from_fsa(removed_fsa_href))
                # Stop polling the departed FSA's Time resource. Left in place
                # these entries are fetched forever: a 404 on a removed FSA is
                # benign to the Time poll (its siblings still refresh), so
                # nothing ever escalates and nothing ever removes them.
                state.fsa_time.pop(removed_fsa_href, None)
                state.fsa_time_hrefs.pop(removed_fsa_href, None)
                client.timebase.forget_fsa(removed_fsa_href)

        # Update snapshot for next poll
        state.previous_fsa_hrefs[edev_href] = current_fsa_hrefs

    logger.debug("Refreshed FSA lists for %d end devices", len(state.end_devices))
    return removed_program_hrefs

refresh_tariffs async

refresh_tariffs(client: Sep2Client, state: DiscoveredState) -> None

Re-walk the tariff tree from all known FSAs (Pricing poll refresh).

Rebuilds state.tariff_profiles so newly-added TimeTariffIntervals are picked up and expired ones drop out. No-op unless pricing is enabled. The walk dedupes within this refresh (the global tariff tree is reachable from every FSA), so each profile is fetched once.

Source code in src/py20305/client/discovery.py

async def refresh_tariffs(client: Sep2Client, state: DiscoveredState) -> None:
    """Re-walk the tariff tree from all known FSAs (Pricing poll refresh).

    Rebuilds ``state.tariff_profiles`` so newly-added TimeTariffIntervals are
    picked up and expired ones drop out. No-op unless pricing is enabled. The
    walk dedupes within this refresh (the global tariff tree is reachable from
    every FSA), so each profile is fetched once.
    """
    if not state.pricing_enabled:
        return
    state.tariff_profiles.clear()
    for edev_state in state.end_devices.values():
        for fsa in edev_state.fsa_list:
            await _discover_tariffs_for_fsa(client, fsa, state)

refresh_der_programs async

refresh_der_programs(client: Sep2Client, state: DiscoveredState) -> list[str]

Re-fetch DER program lists from all known FSAs.

Updates existing programs (primacy, program object) and discovers new programs that have appeared since the last full discovery. New programs are fully discovered (DDERC, DERC list, curves) and added to the device mapping.

Prunes programs that are no longer present in any FSA's DER program list. Programs from FSAs whose DERP list returned 404 are preserved (we can't verify their status).

Returns a list of removed program hrefs so callers can cancel events.

Source code in src/py20305/client/discovery.py

async def refresh_der_programs(client: Sep2Client, state: DiscoveredState) -> list[str]:
    """Re-fetch DER program lists from all known FSAs.

    Updates existing programs (primacy, program object) and discovers new
    programs that have appeared since the last full discovery. New programs
    are fully discovered (DDERC, DERC list, curves) and added to the device
    mapping.

    Prunes programs that are no longer present in any FSA's DER program
    list. Programs from FSAs whose DERP list returned 404 are preserved
    (we can't verify their status).

    Returns a list of removed program hrefs so callers can cancel events.
    """
    if not state.end_devices:
        return []

    # Track which programs are confirmed alive and which FSAs we couldn't verify.
    seen_programs: set[str] = set()
    unreachable_fsas: set[str | None] = set()

    for edev_href, edev_state in list(state.end_devices.items()):
        for fsa in edev_state.fsa_list:
            derp_list_href = _extract_href(fsa.derprogram_list_link)
            if not derp_list_href:
                continue

            try:
                derp_pages = await client.get_list(derp_list_href, DerprogramList)
            except Sep2ProtocolError as exc:
                if exc.status_code == 404:
                    logger.warning("DERP list at %s not found during refresh", derp_list_href)
                    unreachable_fsas.add(fsa.href)
                    continue
                raise

            for derp_page in derp_pages:
                for derp in derp_page.derprogram:
                    derp_href = derp.href
                    if not derp_href:
                        continue

                    seen_programs.add(derp_href)
                    # Outside the branches below: those ask whether the
                    # *program* is new, while the entry that needs creating is
                    # the *(program, device)* pair. A device whose new function
                    # set assignment names a program already known took the
                    # update branch and was never mapped onto it, so it received
                    # none of that program's controls until a full discovery
                    # rebuilt the mapping. Unconditional here matches what full
                    # discovery does, and is safe because `add` admits a pair
                    # once -- appending on every refresh poll would grow the
                    # dispatch target list without bound.
                    state.device_mapping.add(derp_href, edev_href)

                    if derp_href in state.der_programs:
                        # Update the program object and primacy
                        state.der_programs[derp_href].program = derp
                        state.der_programs[derp_href].primacy = (
                            derp.primacy.value if derp.primacy is not None else 0
                        )
                    else:
                        # New program discovered during refresh
                        derp_state = await _discover_program(client, derp, derp_href)
                        derp_state.discovered_from_fsa_href = fsa.href
                        state.der_programs[derp_href] = derp_state
                        logger.info(
                            "Discovered new program %s (primacy=%d) during refresh",
                            derp_href,
                            derp_state.primacy,
                        )

    # Prune programs no longer in any FSA's DERP list. Preserve programs
    # from FSAs we couldn't reach (404) since we can't confirm removal.
    removed: list[str] = []
    for href, ps in list(state.der_programs.items()):
        if href in seen_programs:
            continue
        if ps.discovered_from_fsa_href in unreachable_fsas:
            continue
        logger.info("Removing stale program %s (no longer in any DERP list)", href)
        del state.der_programs[href]
        state.device_mapping.remove_program(href)
        removed.append(href)

    logger.debug(
        "Refreshed DER programs for %d end devices (%d pruned)",
        len(state.end_devices),
        len(removed),
    )
    return removed

py20305.client.retry

Retry logic with exponential backoff for the SEP2 client.

RetryPolicy dataclass

Configuration for retry behavior.

Source code in src/py20305/client/retry.py

@dataclass(frozen=True)
class RetryPolicy:
    """Configuration for retry behavior."""

    max_transient: int = 3
    max_tls: int = 8
    base_delay: float = 0.25
    backoff_factor: float = 2.0

with_retry async

with_retry(policy: RetryPolicy, operation: Callable[..., Awaitable[T]], *args: object, peer: str | None = None, **kwargs: object) -> T

Execute an async operation with retry on transient/TLS errors.

peer is used to tag the diagnostic surfaced when transient retries are exhausted (Sep2ConnectionError). Callers that have it (e.g. the SEP2 HTTP client) should pass it so the UI dedup-collapses one flapping host to a single entry; callers without context can omit it and only the log line will fire.

Source code in src/py20305/client/retry.py

async def with_retry(
    policy: RetryPolicy,
    operation: Callable[..., Awaitable[T]],
    *args: object,
    peer: str | None = None,
    **kwargs: object,
) -> T:
    """Execute an async operation with retry on transient/TLS errors.

    ``peer`` is used to tag the diagnostic surfaced when transient retries
    are exhausted (``Sep2ConnectionError``). Callers that have it (e.g.
    the SEP2 HTTP client) should pass it so the UI dedup-collapses one
    flapping host to a single entry; callers without context can omit it
    and only the log line will fire.
    """
    transient_attempts = 0
    tls_attempts = 0

    while True:
        try:
            return await operation(*args, **kwargs)
        except ssl.SSLError as exc:
            tls_attempts += 1
            hostname_mismatch = _is_hostname_mismatch(exc)
            if hostname_mismatch and tls_attempts == 1:
                # Log the remediation hint once on first encounter; subsequent
                # retries fall through to the generic per-attempt log so the
                # noise stays bounded. By construction, getting this error at
                # all means check_hostname is currently True -- Python's ssl
                # layer doesn't run hostname matching otherwise.
                logger.warning(
                    "TLS hostname mismatch%s: %s. %s",
                    f" to {peer}" if peer else "",
                    exc,
                    _HOSTNAME_MISMATCH_HINT,
                )
            if tls_attempts >= policy.max_tls:
                if peer is not None:
                    from py20305.diagnostics import report

                    if hostname_mismatch:
                        message = (
                            f"TLS hostname mismatch to {peer} after "
                            f"{tls_attempts} attempts: {exc}. "
                            f"{_HOSTNAME_MISMATCH_HINT}"
                        )
                        dedup_key = f"tls_hostname_mismatch:{peer}"
                    else:
                        message = (
                            f"TLS handshake failure to {peer} after {tls_attempts} attempts: {exc}"
                        )
                        dedup_key = f"tls_handshake:{peer}:{type(exc).__name__}"
                    details: dict[str, object] = {
                        "peer": peer,
                        "error_kind": type(exc).__name__,
                        "error": str(exc),
                    }
                    if hostname_mismatch:
                        details["kind"] = "hostname_mismatch"
                    report(
                        "warnings",
                        message,
                        source="client",
                        dedup_key=dedup_key,
                        details=details,
                    )
                raise Sep2TlsError(f"TLS error after {tls_attempts} attempts: {exc}") from exc
            delay = policy.base_delay * (policy.backoff_factor ** (tls_attempts - 1))
            logger.warning(
                "TLS error (attempt %d/%d), retrying in %.2fs: %s",
                tls_attempts,
                policy.max_tls,
                delay,
                exc,
            )
            await asyncio.sleep(delay)
        except Sep2RateLimitError as exc:
            transient_attempts += 1
            if transient_attempts >= policy.max_transient:
                raise
            delay = (
                float(exc.retry_after)
                if exc.retry_after
                else (policy.base_delay * (policy.backoff_factor ** (transient_attempts - 1)))
            )
            logger.warning(
                "Rate limited (attempt %d/%d), retrying in %.2fs",
                transient_attempts,
                policy.max_transient,
                delay,
            )
            await asyncio.sleep(delay)
        except Sep2RedirectError as exc:
            # IEEE 5.5.2.7 says clients SHOULD re-discover on a redirect, but in
            # practice these (301/302/307/308) are usually transient (e.g. a
            # proxy returning `/Error/?errorCode=4&404` because of a malformed
            # upstream request). Retrying with backoff lets a transient cause
            # clear without taking the whole process down. After max_transient
            # attempts the original error propagates so an operator who is
            # actually facing a permanent move sees it; we don't follow the
            # Location blindly because the target may be a cross-origin
            # error page. Following + re-discovery is tracked separately.
            transient_attempts += 1
            if transient_attempts >= policy.max_transient:
                raise
            delay = policy.base_delay * (policy.backoff_factor ** (transient_attempts - 1))
            location = exc.location or "(no Location header)"
            if peer is not None:
                from py20305.diagnostics import report

                # Surfaces the redirect status + target so the operator sees
                # *what* and *where* the upstream is sending us (often the actual
                # diagnostic). Dedup per (peer, status, location) so a stable
                # redirect collapses to one entry; a fresh one still produces a
                # new entry.
                report(
                    "warnings",
                    f"HTTP {exc.status_code} redirect from {peer} -> {location} "
                    f"(attempt {transient_attempts}/{policy.max_transient})",
                    source="client",
                    dedup_key=f"redirect:{peer}:{exc.status_code}:{location}",
                    details={
                        "peer": peer,
                        "status_code": exc.status_code,
                        "location": location,
                        "attempt": transient_attempts,
                    },
                )
            else:
                logger.warning(
                    "HTTP %d redirect (attempt %d/%d) -> %s, retrying in %.2fs",
                    exc.status_code,
                    transient_attempts,
                    policy.max_transient,
                    location,
                    delay,
                )
            await asyncio.sleep(delay)
        except (TimeoutError, OSError) as exc:
            transient_attempts += 1
            if transient_attempts >= policy.max_transient:
                msg = f"Connection error after {transient_attempts} attempts: {exc}"
                if peer is not None:
                    from py20305.diagnostics import report

                    report(
                        "errors",
                        f"Cannot reach {peer}: {exc}",
                        source="client",
                        dedup_key=f"connection:{peer}",
                        details={"peer": peer, "error": str(exc)},
                    )
                raise Sep2ConnectionError(msg) from exc
            delay = policy.base_delay * (policy.backoff_factor ** (transient_attempts - 1))
            logger.warning(
                "Transient error (attempt %d/%d), retrying in %.2fs: %s",
                transient_attempts,
                policy.max_transient,
                delay,
                exc,
            )
            await asyncio.sleep(delay)

py20305.security.identity

LFDI/SFDI identity computation per IEEE 2030.5.

compute_lfdi

compute_lfdi(cert_pem: str) -> str

Compute the LFDI from a PEM-encoded certificate.

The leftmost 40 hex of the certificate's SHA-256 fingerprint, lowercase (per IEEE 2030.5). Derived from :func:compute_cert_fingerprint so the two can never disagree about the underlying digest. Raises ValueError if the certificate cannot be parsed.

Source code in src/py20305/security/identity.py

def compute_lfdi(cert_pem: str) -> str:
    """Compute the LFDI from a PEM-encoded certificate.

    The leftmost 40 hex of the certificate's SHA-256 fingerprint, lowercase
    (per IEEE 2030.5). Derived from :func:`compute_cert_fingerprint` so the two
    can never disagree about the underlying digest. Raises ValueError if the
    certificate cannot be parsed.
    """
    return compute_cert_fingerprint(cert_pem)[:40].lower()

compute_cert_fingerprint

compute_cert_fingerprint(cert_pem: str) -> str

Compute the full SHA-256 fingerprint of a PEM-encoded certificate.

SHA-256 of the DER-encoded certificate, all 64 hex characters, uppercase. This is the value some servers key device/aggregator enrollment on (the LFDI is the leftmost 40 hex of the same digest). Raises ValueError if the certificate cannot be parsed.

Source code in src/py20305/security/identity.py

def compute_cert_fingerprint(cert_pem: str) -> str:
    """Compute the full SHA-256 fingerprint of a PEM-encoded certificate.

    SHA-256 of the DER-encoded certificate, all 64 hex characters, uppercase.
    This is the value some servers key device/aggregator enrollment on (the
    LFDI is the leftmost 40 hex of the same digest). Raises ValueError if the
    certificate cannot be parsed.
    """
    try:
        cert = x509.load_pem_x509_certificate(cert_pem.encode("utf-8"))
    except Exception as exc:
        raise ValueError(f"Failed to parse PEM certificate: {exc}") from exc
    der_bytes = cert.public_bytes(Encoding.DER)
    return hashlib.sha256(der_bytes).hexdigest().upper()

compute_sfdi

compute_sfdi(lfdi: str) -> int

Compute the SFDI from an LFDI string.

Per IEEE 2030.5 §6.3.3: the SFDI is the 36 most-significant bits of the LFDI (its first 9 hex characters) as a decimal, with a sum-of-digits checksum digit appended so the whole SFDI's digit sum is a multiple of 10. This is a sum-of-digits checksum, NOT Luhn -- the two coincide for many values but diverge for others (e.g. LFDI 3e4f45ab3... -> ...391, not ...397), and the server rejects the wrong one.

Source code in src/py20305/security/identity.py

def compute_sfdi(lfdi: str) -> int:
    """Compute the SFDI from an LFDI string.

    Per IEEE 2030.5 §6.3.3: the SFDI is the 36 most-significant bits of the LFDI
    (its first 9 hex characters) as a decimal, with a **sum-of-digits** checksum
    digit appended so the whole SFDI's digit sum is a multiple of 10. This is a
    sum-of-digits checksum, NOT Luhn -- the two coincide for many values but
    diverge for others (e.g. LFDI ``3e4f45ab3...`` -> ``...391``, not ``...397``),
    and the server rejects the wrong one.
    """
    if len(lfdi) < 9:
        raise ValueError(f"LFDI must be at least 9 characters, got {len(lfdi)}")
    base = int(lfdi[:9], 16)
    check = (10 - sum(int(d) for d in str(base)) % 10) % 10
    return base * 10 + check

generate_proxied_lfdi

generate_proxied_lfdi(pen: int = _DEFAULT_PEN) -> str

Generate an LFDI for a proxied device that has no certificate of its own.

A device behind a gateway still needs an LFDI to be addressed by, but has no certificate to derive one from. This builds one instead: 128 random bits concatenated with the PEN, SHA-256 hashed, truncated to 40 hex characters. The PEN makes collisions across vendors improbable, and the randomness makes them improbable within one.

Parameters:

Name	Type	Description	Default
pen	int	IANA Private Enterprise Number to attribute the generated identity to. Pass your own organization's PEN; the default is only a placeholder so the function is callable out of the box.	_DEFAULT_PEN

Returns:

Type	Description
str	40-character lowercase hex LFDI string.

Source code in src/py20305/security/identity.py

def generate_proxied_lfdi(pen: int = _DEFAULT_PEN) -> str:
    """Generate an LFDI for a proxied device that has no certificate of its own.

    A device behind a gateway still needs an LFDI to be addressed by, but has
    no certificate to derive one from. This builds one instead: 128 random
    bits concatenated with the PEN, SHA-256 hashed, truncated to 40 hex
    characters. The PEN makes collisions across vendors improbable, and the
    randomness makes them improbable within one.

    Args:
        pen: IANA Private Enterprise Number to attribute the generated
            identity to. Pass your own organization's PEN; the default is
            only a placeholder so the function is callable out of the box.

    Returns:
        40-character lowercase hex LFDI string.
    """
    random_bytes = secrets.token_bytes(16)
    pen_bytes = pen.to_bytes(4, byteorder="big")
    combined = random_bytes + pen_bytes
    digest = hashlib.sha256(combined).hexdigest()
    return digest[:40]

py20305.config

The configuration file a deployed client is started from.

One document describes a running client: which server to talk to, which certificate to present, which devices to drive, and what to expose locally. It is validated on load, so a mistake surfaces at startup with a message naming the field rather than hours later as a failed poll.

Loading is separate from the models on purpose. Anything embedding this client in its own application already has configuration of its own and should build :class:ClientConfig directly, or skip it and construct the client itself -- :func:load_config exists for the command-line runner, not as the only way in.

ConfigError

Bases: Exception

A configuration file could not be read, parsed, or validated.

Source code in src/py20305/config.py

class ConfigError(Exception):
    """A configuration file could not be read, parsed, or validated."""

ServerConfig

Bases: _Strict

The IEEE 2030.5 server this client talks to.

Source code in src/py20305/config.py

class ServerConfig(_Strict):
    """The IEEE 2030.5 server this client talks to."""

    url: str | None = Field(
        default=None,
        description=(
            "Base URL of the server, e.g. https://server.example.com:8443. Omit it to "
            "have the client locate a server by DNS-SD query at startup, which is what "
            "IEEE 2030.5 §6.9.2 asks a client to do; `discovery` then has to be enabled."
        ),
    )

    dcap_path: str = Field(
        default="/dcap",
        description="Path of the DeviceCapability resource. Standard servers use /dcap.",
    )
    poll_now_on_start: bool = Field(
        default=True,
        description="Poll once immediately after connecting rather than waiting for the schedule.",
    )
    server_2018_compat: bool = Field(
        default=False,
        description="Adjust behavior for servers implementing IEEE 2030.5-2018 rather than 2023.",
    )
    use_server_time: bool = Field(
        default=True,
        description=(
            "Follow the server's Time resource for scheduling rather than the local clock. "
            "The local clock is never modified either way."
        ),
    )
    fsa_stale_seconds: float | None = Field(
        default=None,
        gt=0,
        description=(
            "How old an FSA's own Time observation may get before scheduling for its "
            "programs falls back to the global timebase. Unset derives it from the Time "
            "poll cadence (three polls, at least an hour), which is what a healthy "
            "deployment wants; set it only to override that."
        ),
    )

    @field_validator("url")
    @classmethod
    def _must_be_a_usable_https_url(cls, v: str | None) -> str | None:
        """Reject an unusable URL at load rather than at handshake.

        IEEE 2030.5 is mutual TLS throughout, so a plain http:// URL cannot
        work. The URL is parsed rather than prefix-matched because
        ``https://`` alone passes a prefix check and then fails much later
        inside the HTTP client, outside this module's error handling and
        looking like the server is down.
        """
        if v is None:
            return None
        parsed = urlparse(v)
        if parsed.scheme != "https":
            raise ValueError(f"must be an https:// URL (IEEE 2030.5 requires TLS), got {v!r}")
        if not parsed.hostname:
            raise ValueError(f"must include a host, got {v!r}")
        return v.rstrip("/")

DiscoveryConfig

Bases: _Strict

Locate an IEEE 2030.5 server by DNS-SD query, per Clause 7.

On by default because the standard puts it on the client: §6.9.2 says "Clients SHALL locate local services by performing DNS service discovery (DNS-SD) queries to the local network." It only runs when no server.url is configured, though, because §7.6 a) lists "use known URI(s) to DeviceCapability resource(s) of interest" as an equally valid way to find a server, and an operator who named one has already answered the question.

Retrying a query that found nothing is governed by the connection block, the same as retrying a connection that failed. A client that starts before its server is on the network and a client whose server is briefly down are the same situation to whoever set that policy.

Source code in src/py20305/config.py

class DiscoveryConfig(_Strict):
    """Locate an IEEE 2030.5 server by DNS-SD query, per Clause 7.

    On by default because the standard puts it on the client: §6.9.2 says
    "Clients SHALL locate local services by performing DNS service discovery
    (DNS-SD) queries to the local network." It only runs when no `server.url`
    is configured, though, because §7.6 a) lists "use known URI(s) to
    DeviceCapability resource(s) of interest" as an equally valid way to find
    a server, and an operator who named one has already answered the question.

    Retrying a query that found nothing is governed by the `connection` block,
    the same as retrying a connection that failed. A client that starts before
    its server is on the network and a client whose server is briefly down are
    the same situation to whoever set that policy.
    """

    enabled: bool = Field(
        default=True,
        description=(
            "Locate a server by DNS-SD query when no server.url is set. Turning this "
            "off makes server.url required."
        ),
    )
    transport: Literal["mdns", "xmdns", "both"] = Field(
        default="mdns",
        description=(
            "Which multicast transport to query over. mdns is normative in "
            "IEEE 2030.5-2023 (.local); xmdns is the 2018 transport (.site) and is "
            "deprecated; both queries each in turn."
        ),
    )
    timeout_seconds: float = Field(
        default=3.0,
        gt=0,
        le=60,
        description="How long to listen for answers, per transport.",
    )
    subtype: str | None = Field(
        default=None,
        description=(
            "Narrow the query to one function set using a IEEE 2030.5 §7.5 Table 17 "
            "subtype, such as `derp`. Unset asks for the client's own EndDevice first "
            "and then for any server, which is the sequence Annex C describes."
        ),
    )
    interface: str | None = Field(
        default=None,
        description=(
            "Interface to query from -- an address for IPv4, a name for IPv6. Unset "
            "uses the route to the multicast group, which on a gateway with both a "
            "utility uplink and a device LAN may not be the intended one."
        ),
    )
    @field_validator("subtype")
    @classmethod
    def _subtype_is_one_label(cls, v: str | None) -> str | None:
        """§7.5: a subtype string is one DNS label and does not start with `_`."""
        if v is None:
            return v
        try:
            return validate_subtype(v)
        except ValueError as exc:
            raise ValueError(str(exc)) from exc

AdvertiseConfig

Bases: _Strict

Announce this client on the local network over multicast DNS-SD.

IEEE 2030.5 §7.1 defines two multicast transports and the editions differ over which is normative: the 2018 edition on xmDNS (site-local FF05::FB, the .site domain) and the 2023 edition on plain mDNS (link-local, .local), where xmDNS is retained but deprecated. The records themselves are identical either way, so the choice is a transport rather than a format.

Source code in src/py20305/config.py

class AdvertiseConfig(_Strict):
    """Announce this client on the local network over multicast DNS-SD.

    IEEE 2030.5 §7.1 defines two multicast transports and the editions differ
    over which is normative: the 2018 edition on xmDNS (site-local
    ``FF05::FB``, the ``.site`` domain) and the 2023 edition on plain mDNS
    (link-local, ``.local``), where xmDNS is retained but deprecated. The
    records themselves are identical either way, so the choice is a transport
    rather than a format.
    """

    transport: Literal["mdns", "xmdns", "both", "off"] = Field(
        default="off",
        description=(
            "Which multicast transport to announce on. mdns uses the .local domain "
            "and xmdns the .site one, both announces on each, and off announces "
            "nothing. Off by default: announcing is not part of IEEE 2030.5, and it "
            "publishes this client's LFDI and SFDI to everything on the segment."
        ),
    )
    service: str = Field(
        default=DEFAULT_SERVICE,
        description=(
            "The DNS-SD service to announce under. The default is deliberately not "
            "the registered _smartenergy._tcp: this client is not an IEEE 2030.5 "
            "server, and claiming that name would have conformant clients try to "
            "read a DeviceCapability resource it does not serve."
        ),
    )
    port: int | None = Field(
        default=None,
        gt=0,
        le=65535,
        description=(
            "The TCP port to advertise. Unset advertises the management API when it "
            "is enabled, otherwise the notification server, and announces nothing "
            "when neither is -- a service record has to point at something."
        ),
    )
    instance: str | None = Field(
        default=None,
        description=(
            "Override the instance name. The default ends with this client's SFDI, "
            "which IEEE 2030.5 §7.2 gives as the way to keep the name unique; set "
            "this only when two clients share one certificate."
        ),
    )
    interface: str | None = Field(
        default=None,
        description=(
            "Interface to announce from -- an address for IPv4, a name for IPv6. "
            "Unset uses the route to the multicast group, which on a gateway with "
            "both a utility uplink and a device LAN may not be the intended one."
        ),
    )
    txt: dict[str, str] = Field(
        default_factory=dict,
        description="Additional TXT record keys to publish alongside the defaults.",
    )

    @field_validator("service")
    @classmethod
    def _must_be_a_dns_sd_service(cls, v: str) -> str:
        """Reject a malformed service name at load rather than at announcement.

        Deferring it would surface as a name nothing ever answers, which looks
        like a network fault rather than a typo in this field.
        """
        try:
            validate_service(v)
        except ValueError as exc:
            raise ValueError(str(exc)) from exc
        return v

    @field_validator("instance")
    @classmethod
    def _instance_fits_a_label(cls, v: str | None) -> str | None:
        """§7.2: an instance label is up to 63 bytes of UTF-8."""
        if v is None:
            return v
        if not v:
            raise ValueError("instance must not be empty")
        if len(v.encode("utf-8")) > 63:
            raise ValueError(f"instance {v!r} is longer than 63 bytes (IEEE 2030.5 §7.2)")
        return v

TlsFileConfig

Bases: _Strict

The certificate material this client presents and trusts.

Paths are resolved relative to the configuration file, so a deployment can be moved without rewriting them.

Source code in src/py20305/config.py

class TlsFileConfig(_Strict):
    """The certificate material this client presents and trusts.

    Paths are resolved relative to the configuration file, so a deployment can
    be moved without rewriting them.
    """

    client_cert: Path = Field(description="Client certificate (PEM). Its LFDI identifies you.")
    client_key: Path = Field(description="Private key for the client certificate (PEM).")
    ca_cert: Path = Field(description="CA bundle that signed the server's certificate (PEM).")

    check_hostname: bool = Field(
        default=True,
        description=(
            "Verify the server's certificate names the host being connected to. "
            "Turn off only for a server reached by an address its certificate does not carry."
        ),
    )
    additional_ciphers: tuple[str, ...] = Field(
        default=(),
        description=(
            "Extra TLS 1.2 cipher suites, each naming exactly one suite. For peers presenting "
            "RSA certificates, which the IEEE 2030.5 ECDSA-only baseline cannot handshake with."
        ),
    )

SubscriptionConfig

Bases: _Strict

Subscribe/notify instead of relying on polling alone.

Off by default: enabling it opens a listening socket for the server's notifications, which a deployment should choose rather than inherit.

Source code in src/py20305/config.py

class SubscriptionConfig(_Strict):
    """Subscribe/notify instead of relying on polling alone.

    Off by default: enabling it opens a listening socket for the server's
    notifications, which a deployment should choose rather than inherit.
    """

    enabled: bool = Field(
        default=False,
        description="Subscribe to server resources and receive notifications",
    )
    notification_host: str = Field(
        default="0.0.0.0",
        description="Address the notification listener binds",
    )
    notification_port: int = Field(
        default=10443,
        ge=1,
        le=65535,
        description="Port the notification listener binds",
    )
    notification_external_host: str | None = Field(
        default=None,
        description=(
            "Hostname or IP the server should deliver notifications to -- what goes "
            "into each subscription's notificationURI. Required when enabled: the "
            "bind address is not it, and advertising 0.0.0.0 would subscribe with a "
            "callback no server can reach."
        ),
    )
    notification_client_cert_mode: Literal["off", "warn", "enforce"] = Field(
        default="warn",
        description="How strictly the listener checks the notifying server's client certificate",
    )

    @model_validator(mode="after")
    def _external_host_when_enabled(self) -> SubscriptionConfig:
        """The advertised callback host cannot be guessed, so it must be stated."""
        if self.enabled and not self.notification_external_host:
            raise ValueError(
                "subscription.enabled requires subscription.notification_external_host -- "
                "it becomes the notificationURI the server delivers to"
            )
        return self

TelemetryConfig

Bases: _Strict

Reporting each device's measurements and state back to the server.

On by default: a utility program expects the devices in it to report, and a client that registers and then says nothing looks to the server like one that has failed. Set enabled: false for a deployment that should observe and dispatch without reporting.

One switch covers both halves, because they are one conversation with the utility: readings mirrored as MirrorUsagePoints, and the DER resources (DERCapability, DERSettings, DERStatus, DERAvailability) PUT for each device. A server exposing no MirrorUsagePointList disables the readings only -- it still expects the DER resources.

Source code in src/py20305/config.py

class TelemetryConfig(_Strict):
    """Reporting each device's measurements and state back to the server.

    On by default: a utility program expects the devices in it to report, and a
    client that registers and then says nothing looks to the server like one
    that has failed. Set ``enabled: false`` for a deployment that should
    observe and dispatch without reporting.

    One switch covers both halves, because they are one conversation with the
    utility: readings mirrored as MirrorUsagePoints, and the DER resources
    (DERCapability, DERSettings, DERStatus, DERAvailability) PUT for each
    device. A server exposing no MirrorUsagePointList disables the readings
    only -- it still expects the DER resources.
    """

    enabled: bool = Field(
        default=True,
        description="Read each configured device on a schedule and report it to the server",
    )
    post_rate_seconds: int = Field(
        default=300,
        gt=0,
        description=(
            "How often each device is read, its readings posted and its DERStatus PUT. "
            "The server's EndDevice.postRate takes precedence when it specifies one."
        ),
    )
    der_capability_poll_rate_seconds: int = Field(
        default=86400,
        gt=0,
        description="How often DERCapability is PUT. Nameplate data, so a day by default.",
    )
    der_settings_poll_rate_seconds: int = Field(
        default=60,
        gt=0,
        description="How often DERSettings is PUT, when it has changed",
    )
    reading_profile: Literal["default", "signed_load_convention"] = Field(
        default="default",
        description=(
            "How MirrorMeterReadings are encoded. 'signed_load_convention' carries the "
            "sign on each value (flowDirection 0) with export and reactive power "
            "injection negative, and adds line-to-line voltage and State of Energy "
            "readings when the device supplies them. Applied at startup."
        ),
    )

LoggingConfig

Bases: _Strict

How much the runner says while it works.

Source code in src/py20305/config.py

class LoggingConfig(_Strict):
    """How much the runner says while it works."""

    level: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "INFO"
    file: Path | None = Field(
        default=None,
        description="Also write to this file. Logs go to stderr regardless.",
    )

ApiConfig

Bases: _Strict

The optional local management API.

Source code in src/py20305/config.py

class ApiConfig(_Strict):
    """The optional local management API."""

    enabled: bool = False
    host: str = Field(
        default="127.0.0.1",
        description=(
            "Bind address. Loopback by default: the API is unauthenticated, so exposing it "
            "on a routable address makes the client's controls available to that network."
        ),
    )
    port: int = Field(default=8080, ge=1, le=65535)

ConnectionConfig

Bases: _Strict

What to do when the server cannot be reached.

A client on a gateway is expected to outlive the server's outages, so it retries indefinitely by default rather than exiting and relying on something else to restart it.

Source code in src/py20305/config.py

class ConnectionConfig(_Strict):
    """What to do when the server cannot be reached.

    A client on a gateway is expected to outlive the server's outages, so it
    retries indefinitely by default rather than exiting and relying on
    something else to restart it.
    """

    retry_forever: bool = Field(default=True, description="Keep retrying the initial connection.")
    max_attempts: int = Field(
        default=0,
        ge=0,
        description="Give up after this many attempts. 0 means never, and requires retry_forever.",
    )
    initial_delay_seconds: float = Field(default=5.0, gt=0)
    max_delay_seconds: float = Field(default=300.0, gt=0)
    backoff_factor: float = Field(default=2.0, ge=1.0)

ClientConfig

Bases: _Strict

A complete description of one running client.

Source code in src/py20305/config.py

class ClientConfig(_Strict):
    """A complete description of one running client."""

    server: ServerConfig
    tls: TlsFileConfig
    devices: list[Annotated[DeviceConfig, Field(discriminator="type")]] = Field(
        default_factory=list,
        description="The devices this client drives. An empty list monitors without dispatching.",
    )

    logging: LoggingConfig = Field(default_factory=LoggingConfig)
    api: ApiConfig = Field(default_factory=ApiConfig)
    connection: ConnectionConfig = Field(default_factory=ConnectionConfig)
    telemetry: TelemetryConfig = Field(default_factory=TelemetryConfig)
    subscription: SubscriptionConfig = Field(default_factory=SubscriptionConfig)
    discovery: DiscoveryConfig = Field(default_factory=DiscoveryConfig)
    advertise: AdvertiseConfig = Field(default_factory=AdvertiseConfig)
    forwarders: ForwarderConfig | None = Field(
        default=None,
        description=(
            "Publish captured traffic to a monitoring system. Omitted, the client "
            "forwards nothing and never loads the transport."
        ),
    )

    registration_pins: dict[str, int] = Field(
        default_factory=dict,
        description=(
            "Expected Registration PIN per device LFDI (40 hex characters). When a "
            "device's Registration resource carries a different pIN, the mismatch is "
            "reported at discovery -- the check a utility's registration flow expects."
        ),
    )
    register_on_start: bool = Field(
        default=True,
        description=(
            "Register an EndDevice for this client's own certificate identity if the server "
            "does not already have one. Registering when it does would create a duplicate."
        ),
    )

    @field_validator("registration_pins")
    @classmethod
    def _pins_keyed_by_lfdi(cls, pins: dict[str, int]) -> dict[str, int]:
        """A mistyped key would silently verify nothing."""
        for lfdi in pins:
            if len(lfdi) != 40 or any(c not in "0123456789abcdefABCDEF" for c in lfdi):
                raise ValueError(f"registration_pins key {lfdi!r} is not a 40-hex LFDI")
        return {k.lower(): v for k, v in pins.items()}

    @field_validator("devices")
    @classmethod
    def _reject_duplicate_lfdis(cls, devices: list[Any]) -> list[Any]:
        """Two entries for one device is a configuration error, not a merge.

        The registry keys on LFDI, so a duplicate silently wins over the other
        and the operator sees one of their two devices never being polled.
        """
        seen: set[str] = set()
        for device in devices:
            lfdi = device.lfdi.lower()
            if lfdi in seen:
                raise ValueError(f"duplicate device lfdi {device.lfdi!r}")
            seen.add(lfdi)
        return devices

    @model_validator(mode="after")
    def _a_server_has_to_be_findable(self) -> ClientConfig:
        """Either name a server or leave discovery able to find one.

        With neither, the client has no way to reach anything and would fail
        at the first connect attempt, which reads as an unreachable server
        rather than as the configuration gap it is.
        """
        if self.server.url is None and not self.discovery.enabled:
            raise ValueError(
                "set server.url, or leave discovery.enabled on so the client can "
                "locate a server by DNS-SD query (IEEE 2030.5 §6.9.2)"
            )
        return self

    def resolve_paths(self, base: Path) -> ClientConfig:
        """Return a copy with relative certificate paths resolved against ``base``.

        Relative to the configuration file rather than the working directory,
        so the same deployment works whichever directory it is started from --
        which is not the same directory under systemd as it is by hand.
        """
        tls = self.tls.model_copy(
            update={
                field: (base / value).resolve()
                for field in ("client_cert", "client_key", "ca_cert")
                if not (value := getattr(self.tls, field)).is_absolute()
            }
        )
        logging_config = self.logging
        if logging_config.file is not None and not logging_config.file.is_absolute():
            logging_config = logging_config.model_copy(
                update={"file": (base / logging_config.file).resolve()}
            )

        # A device speaking Modbus over TLS carries its own certificate paths,
        # and they go straight to an SSL context. Resolving only the server's
        # would make the same relative path mean different things depending on
        # which section it appeared in.
        devices = [self._resolve_device_paths(device, base) for device in self.devices]

        # The forwarder's paths are paths in the same file, read the same way,
        # so they resolve the same way: its schema directory, and the broker
        # credentials, which go straight to an SSL context exactly as a
        # device's do.
        forwarders = self._resolve_forwarder_paths(self.forwarders, base)

        return self.model_copy(
            update={
                "tls": tls,
                "logging": logging_config,
                "devices": devices,
                "forwarders": forwarders,
            }
        )

    @staticmethod
    def _resolve_forwarder_paths(forwarders: Any, base: Path) -> Any:
        """Resolve the forwarder's schema directory and broker credentials."""
        if forwarders is None:
            return None

        updates: dict[str, Any] = {}
        if forwarders.schema_dir is not None and not forwarders.schema_dir.is_absolute():
            updates["schema_dir"] = (base / forwarders.schema_dir).resolve()

        mqtt = forwarders.mqtt
        if mqtt is not None:
            mqtt_updates = {
                field: (base / value).resolve()
                for field in ("cert_path", "key_path", "ca_path")
                if (value := getattr(mqtt, field, None)) is not None and not value.is_absolute()
            }
            if mqtt_updates:
                updates["mqtt"] = mqtt.model_copy(update=mqtt_updates)

        return forwarders.model_copy(update=updates) if updates else forwarders

    @staticmethod
    def _resolve_device_paths(device: Any, base: Path) -> Any:
        """Resolve a device's own certificate paths, if it has any."""
        updates = {
            field: str((base / value).resolve())
            for field in ("ca_path", "cert_path", "key_path")
            if (value := getattr(device, field, None)) and not Path(value).is_absolute()
        }
        return device.model_copy(update=updates) if updates else device

resolve_paths

resolve_paths(base: Path) -> ClientConfig

Return a copy with relative certificate paths resolved against base.

Relative to the configuration file rather than the working directory, so the same deployment works whichever directory it is started from -- which is not the same directory under systemd as it is by hand.

Source code in src/py20305/config.py

def resolve_paths(self, base: Path) -> ClientConfig:
    """Return a copy with relative certificate paths resolved against ``base``.

    Relative to the configuration file rather than the working directory,
    so the same deployment works whichever directory it is started from --
    which is not the same directory under systemd as it is by hand.
    """
    tls = self.tls.model_copy(
        update={
            field: (base / value).resolve()
            for field in ("client_cert", "client_key", "ca_cert")
            if not (value := getattr(self.tls, field)).is_absolute()
        }
    )
    logging_config = self.logging
    if logging_config.file is not None and not logging_config.file.is_absolute():
        logging_config = logging_config.model_copy(
            update={"file": (base / logging_config.file).resolve()}
        )

    # A device speaking Modbus over TLS carries its own certificate paths,
    # and they go straight to an SSL context. Resolving only the server's
    # would make the same relative path mean different things depending on
    # which section it appeared in.
    devices = [self._resolve_device_paths(device, base) for device in self.devices]

    # The forwarder's paths are paths in the same file, read the same way,
    # so they resolve the same way: its schema directory, and the broker
    # credentials, which go straight to an SSL context exactly as a
    # device's do.
    forwarders = self._resolve_forwarder_paths(self.forwarders, base)

    return self.model_copy(
        update={
            "tls": tls,
            "logging": logging_config,
            "devices": devices,
            "forwarders": forwarders,
        }
    )

load_config

load_config(path: str | Path) -> ClientConfig

Load, parse and validate a configuration file.

Accepts YAML (.yaml/.yml) or JSON (.json). Certificate paths are resolved relative to the file.

Raises:

Type	Description
ConfigError	If the file is missing, unparseable, or invalid. The message names the file and, for a validation failure, the field -- an operator reading a startup log should not have to map a pydantic traceback back onto their document.

Source code in src/py20305/config.py

def load_config(path: str | Path) -> ClientConfig:
    """Load, parse and validate a configuration file.

    Accepts YAML (``.yaml``/``.yml``) or JSON (``.json``). Certificate paths
    are resolved relative to the file.

    Raises:
        ConfigError: If the file is missing, unparseable, or invalid. The
            message names the file and, for a validation failure, the field --
            an operator reading a startup log should not have to map a
            pydantic traceback back onto their document.
    """
    path = Path(path)
    if not path.is_file():
        raise ConfigError(f"configuration file not found: {path}")

    suffix = path.suffix.lower()
    try:
        text = path.read_text(encoding="utf-8")
    except OSError as exc:
        raise ConfigError(f"could not read {path}: {exc}") from exc

    if suffix in _YAML_SUFFIXES:
        raw = _parse_yaml(text, path)
    elif suffix in _JSON_SUFFIXES:
        try:
            raw = json.loads(text)
        except json.JSONDecodeError as exc:
            raise ConfigError(f"{path} is not valid JSON: {exc}") from exc
    else:
        known = ", ".join(sorted(_YAML_SUFFIXES | _JSON_SUFFIXES))
        raise ConfigError(
            f"unsupported configuration format {suffix!r} for {path} (expected {known})"
        )

    if not isinstance(raw, dict):
        raise ConfigError(f"{path} must contain a mapping at the top level")

    try:
        config = ClientConfig.model_validate(raw)
    except ValidationError as exc:
        raise ConfigError(f"invalid configuration in {path}:\n{_describe(exc)}") from exc

    return config.resolve_paths(path.parent.resolve())

py20305.cli

Run a client from a configuration file.

py20305 --config client.yaml

What this adds over constructing a :class:~py20305.client.csip_client.CsipClient yourself is the part a deployment needs and a library should not assume: reading configuration from disk, configuring logging, retrying the initial connection instead of exiting when the server is briefly unavailable, and shutting down cleanly when the supervisor asks.

Deliberately not a supervisor itself. It retries the connection, but a crashed process is left for systemd, Docker or whatever else is already responsible for restarting it -- reimplementing that here would be a worse version of a solved problem, and one that hides failures from the thing meant to observe them.

ConnectionState dataclass

What the runner is doing about the upstream connection.

Published on the app so the management API can report "disconnected, retrying" rather than simply failing to answer. That is the state an operator most wants to see, and the API is least useful if it is only available once the thing it reports on is already working.

Source code in src/py20305/cli.py

@dataclass
class ConnectionState:
    """What the runner is doing about the upstream connection.

    Published on the app so the management API can report "disconnected,
    retrying" rather than simply failing to answer. That is the state an
    operator most wants to see, and the API is least useful if it is only
    available once the thing it reports on is already working.
    """

    phase: str = "connecting"
    detail: str | None = None
    attempts: int = 0
    retry_in_seconds: float | None = None

    def to_dict(self) -> dict[str, Any]:
        return {
            "phase": self.phase,
            "detail": self.detail,
            "attempts": self.attempts,
            "retry_in_seconds": self.retry_in_seconds,
        }

setup_logging

setup_logging(config: LoggingConfig) -> None

Configure logging for the runner.

Called only from here, never from library code, which uses getLogger and leaves handler policy to whoever is embedding it. A library that configures the root logger takes that decision away from its host.

Source code in src/py20305/cli.py

def setup_logging(config: LoggingConfig) -> None:
    """Configure logging for the runner.

    Called only from here, never from library code, which uses ``getLogger``
    and leaves handler policy to whoever is embedding it. A library that
    configures the root logger takes that decision away from its host.
    """
    handlers: list[logging.Handler] = [logging.StreamHandler(sys.stderr)]
    if config.file is not None:
        config.file.parent.mkdir(parents=True, exist_ok=True)
        handlers.append(logging.FileHandler(config.file, encoding="utf-8"))

    logging.basicConfig(
        level=getattr(logging, config.level),
        format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
        handlers=handlers,
        force=True,
    )
    # aiohttp logs every request at INFO, which buries this client's own
    # messages under traffic the operator did not ask to see.
    logging.getLogger("aiohttp.access").setLevel(logging.WARNING)

find_servers

find_servers(config: ClientConfig, own_sfdi: int | None) -> list[DiscoveredServer]

Query the local network for IEEE 2030.5 servers (IEEE 2030.5 §6.9.2).

Blocking, and called from a thread by the runner. The sockets underneath are plain blocking sockets on purpose: multicast behavior is the part that differs most between platforms, and keeping it out of the event loop keeps those differences away from the protocol client.

A configured subtype takes precedence over the SFDI sequence rather than suppressing it: the subtype goes into the PTR name, so an operator who named one asked a specific question and should get an answer to it.

Source code in src/py20305/cli.py

def find_servers(config: ClientConfig, own_sfdi: int | None) -> list[DiscoveredServer]:
    """Query the local network for IEEE 2030.5 servers (IEEE 2030.5 §6.9.2).

    Blocking, and called from a thread by the runner. The sockets underneath
    are plain blocking sockets on purpose: multicast behavior is the part that
    differs most between platforms, and keeping it out of the event loop keeps
    those differences away from the protocol client.

    A configured subtype takes precedence over the SFDI sequence rather than
    suppressing it: the subtype goes into the PTR name, so an operator who
    named one asked a specific question and should get an answer to it.
    """
    return discover_for_client(
        transports_for(config.discovery.transport),
        sfdi=own_sfdi,
        subtype=config.discovery.subtype,
        timeout=config.discovery.timeout_seconds,
        interface=config.discovery.interface,
    )

build_client

build_client(config: ClientConfig) -> tuple[CsipClient, str]

Construct a client from configuration, returning it and its own LFDI.

Source code in src/py20305/cli.py

def build_client(config: ClientConfig) -> tuple[CsipClient, str]:
    """Construct a client from configuration, returning it and its own LFDI."""
    # Resolved by discovery before this is reached whenever it was not
    # configured, so an unset URL here is a wiring mistake rather than an
    # operator one, and should not be reported as a connection failure.
    if config.server.url is None:
        raise ValueError("server.url was not resolved before the client was built")

    tls = TlsConfig(
        client_cert=config.tls.client_cert,
        client_key=config.tls.client_key,
        ca_cert=config.tls.ca_cert,
        check_hostname=config.tls.check_hostname,
        additional_ciphers=config.tls.additional_ciphers,
    )
    own_lfdi = compute_lfdi(config.tls.client_cert.read_text())

    registry = ConnectorConfigRegistry(config.devices)

    sole_device = config.devices[0].lfdi if len(config.devices) == 1 else None
    holder: dict[str, CsipClient] = {}
    resolve_lfdi = make_lfdi_resolver(lambda: holder.get("client"), sole_device)

    # Built before the client, because the dispatcher takes the emitter and the
    # client takes the dispatcher. Both are None when nothing is configured,
    # and every path downstream of them is a no-op in that case.
    forwarder = build_forwarder(config, own_lfdi)
    telemetry = (
        DeviceTelemetryEmitter(
            forwarder,
            config.forwarders.device_telemetry,
            client_id=own_lfdi,
            audit=config.forwarders.audit,
        )
        if config.forwarders is not None
        else None
    )
    # Enabled with nowhere to publish. Device telemetry rides the forwarder's
    # transport rather than owning one, so this configuration silently records
    # nothing -- which looks exactly like a device that is never read or
    # written, and is worth saying out loud.
    if config.forwarders is not None and forwarder is None:
        for name, section in (
            ("device telemetry", config.forwarders.device_telemetry),
            ("the audit trail", config.forwarders.audit),
        ):
            if section.enabled:
                logger.warning(
                    "%s is enabled but no forwarder is configured or enabled; nothing "
                    "will be published. Configure `forwarders.mqtt` alongside it.",
                    name,
                )

    client = CsipClient(
        config.server.url,
        tls=tls,
        dispatcher=ConnectorDispatcher(
            registry, lfdi_resolver=resolve_lfdi, telemetry=telemetry
        ),
        dcap_path=config.server.dcap_path,
        registration_pins=config.registration_pins or None,
        server_2018_compat=config.server.server_2018_compat,
        use_server_time=config.server.use_server_time,
        fsa_stale_seconds=config.server.fsa_stale_seconds,
    )
    # The transport reads this to publish each exchange, and `run` reads it
    # back to start and stop the transport -- so the manager needs no second
    # channel out of here and this function keeps its shape.
    client.http.forwarder = forwarder
    if config.forwarders is not None:
        client.audit.attach_forwarder(forwarder)
        client.audit.configure(config.forwarders.audit, client_id=own_lfdi)
        if telemetry is not None:
            # One counter for the client's writes and its lifecycle records.
            telemetry.configure(config.forwarders.device_telemetry, sequence=client.audit.sequence)

    # Connection telemetry: the client's own connection outcomes, on their own
    # topic. Attached through the observer seam so the client stays ignorant
    # of the forwarder package; the emitter rides the same transport as the
    # protocol messages.
    if config.forwarders is not None and config.forwarders.connection_telemetry.enabled:
        if forwarder is None:
            # Same silent-nothing failure mode as device telemetry above: an
            # enabled channel with no transport records nothing and looks like
            # a client that never connects.
            logger.warning(
                "connection telemetry is enabled but no forwarder is configured or "
                "enabled; nothing will be published. Configure `forwarders.mqtt` "
                "alongside it."
            )
        else:
            from py20305.forwarders.connection_telemetry import ConnectionTelemetryEmitter

            emitter = ConnectionTelemetryEmitter(
                forwarder,
                config.forwarders.connection_telemetry,
                product_version=get_package_version(),
            )
            parsed = urlparse(config.server.url)
            emitter.set_server(
                parsed.hostname,
                parsed.port or (443 if parsed.scheme == "https" else 80),
                base_url=config.server.url,
            )
            client.http.connection_observer = emitter
            # The forwarder reports its own broker connection through the same
            # emitter. Attached before the forwarder starts, so the first
            # connection attempt is reported too.
            forwarder.connection_observer = emitter

    # Subscribe/notify. Attached after construction because the manager needs
    # the client's transport, which exists only once the client does. The
    # client starts and stops the listener with its own lifecycle.
    if config.subscription.enabled:
        from py20305.forwarders.connection_telemetry import ConnectionTelemetryEmitter
        from py20305.subscription.manager import SubscriptionManager
        from py20305.subscription.notification_server import NotificationServer

        notification_server = NotificationServer(
            host=config.subscription.notification_host,
            port=config.subscription.notification_port,
            tls=tls,
            client_cert_mode=config.subscription.notification_client_cert_mode,
        )
        # Inbound notification requests are connections too; report them
        # through the same emitter when connection telemetry is on.
        if isinstance(client.http.connection_observer, ConnectionTelemetryEmitter):
            notification_server.connection_observer = client.http.connection_observer
        # The validator guarantees external_host when enabled.
        external_host = config.subscription.notification_external_host
        assert external_host is not None
        manager = SubscriptionManager(
            client=client.http,
            notification_uri=notification_server.build_notification_uri(external_host),
            server_2018_compat=config.server.server_2018_compat,
            # A renewal answered 401/404 means the server has forgotten us;
            # rediscovery re-subscribes against its current state. Wrapped to
            # drop the bool trigger_rediscovery returns, which the manager's
            # callback contract does not carry.
            on_subscription_lost=_rediscover_quietly(client),
        )
        client.attach_subscriptions(manager, notification_server)
        logger.info(
            "subscriptions on; notifications delivered to %s",
            notification_server.build_notification_uri(external_host),
        )
    # Configuring a schema directory has to actually turn validation on.
    # Without this the validator stays unset and every forwarded frame is
    # reported valid, which is worse than not offering the setting at all.
    if config.forwarders is not None and config.forwarders.schema_dir is not None:
        client.http.set_schema_validator(str(config.forwarders.schema_dir))
    holder["client"] = client
    return client, own_lfdi

build_advertiser

build_advertiser(config: ClientConfig) -> MulticastAdvertiser | None

Build the local-network advertiser, or None when nothing is announced.

Returns None both when the operator turned announcement off and when there is no reachable local service to name. A DNS-SD service record has to carry a port, and pointing one at a port nothing answers on advertises a connection that will always be refused -- worse than staying quiet, because it is a fault the finder has to diagnose rather than an absence they can see.

The certificate is read only once both of those have passed, so a client that announces nothing -- the default -- never touches it here.

Source code in src/py20305/cli.py

def build_advertiser(config: ClientConfig) -> MulticastAdvertiser | None:
    """Build the local-network advertiser, or None when nothing is announced.

    Returns None both when the operator turned announcement off and when there
    is no reachable local service to name. A DNS-SD service record has to carry
    a port, and pointing one at a port nothing answers on advertises a
    connection that will always be refused -- worse than staying quiet, because
    it is a fault the finder has to diagnose rather than an absence they can
    see.

    The certificate is read only once both of those have passed, so a client
    that announces nothing -- the default -- never touches it here.
    """
    transports = transports_for(config.advertise.transport)
    if not transports:
        return None

    port = _advertised_port(config)
    if port is None:
        return None

    own_lfdi = compute_lfdi(config.tls.client_cert.read_text())
    advertisement = build_advertisement(
        lfdi=own_lfdi,
        sfdi=compute_sfdi(own_lfdi),
        port=port,
        service=config.advertise.service,
        instance=config.advertise.instance,
        version=get_package_version(),
        extra_txt=config.advertise.txt,
    )
    return MulticastAdvertiser(
        advertisement, transports, interface=config.advertise.interface
    )

build_forwarder

build_forwarder(config: ClientConfig, own_lfdi: str) -> ForwarderManager | None

Assemble the forwarder transport from configuration.

Returns None when nothing is configured or every configured forwarder is disabled, so a deployment that does not forward never starts a broker connection and never imports the MQTT client.

Parameters:

Name	Type	Description	Default
config	ClientConfig	The loaded configuration.	required
own_lfdi	str	This client's own LFDI, used to attribute an exchange that names no device.	required

Source code in src/py20305/cli.py

def build_forwarder(config: ClientConfig, own_lfdi: str) -> ForwarderManager | None:
    """Assemble the forwarder transport from configuration.

    Returns None when nothing is configured or every configured forwarder is
    disabled, so a deployment that does not forward never starts a broker
    connection and never imports the MQTT client.

    Args:
        config: The loaded configuration.
        own_lfdi: This client's own LFDI, used to attribute an exchange that
            names no device.
    """
    forwarders = config.forwarders
    if forwarders is None or not forwarders.has_enabled_forwarders():
        return None

    from py20305.forwarders import ForwarderManager, MQTTForwarder, MQTTForwarderAdapter

    manager = ForwarderManager()
    if forwarders.mqtt is not None and forwarders.mqtt.enabled:
        manager.add_forwarder(MQTTForwarderAdapter(MQTTForwarder(forwarders.mqtt)))
        logger.info(
            "forwarding to %s:%s under %s",
            forwarders.mqtt.endpoint,
            forwarders.mqtt.port,
            forwarders.mqtt.topic_base,
        )
    # Set after the forwarders are registered so it propagates to each of them.
    manager.client_lfdi = own_lfdi
    return manager

make_lfdi_resolver

make_lfdi_resolver(get_client: Callable[[], CsipClient | None], sole_device: str | None) -> Callable[[str], str | None]

Map a server-side EndDevice href to the LFDI of the device it addresses.

A control names a device by href; the connector registry is keyed by LFDI. The discovered EndDevice carries both, so the mapping comes from what the server said rather than from a guess. Returning one configured LFDI for every href would mean that with two or more devices every control resolves to the wrong connector, or to none, and is silently never applied.

sole_device is the fallback for the single-device case, where the mapping is unambiguous and discovery may not have run yet. With several devices there is no safe fallback, so an unknown href resolves to nothing and the dispatcher reports it rather than guessing.

Source code in src/py20305/cli.py

def make_lfdi_resolver(
    get_client: Callable[[], CsipClient | None], sole_device: str | None
) -> Callable[[str], str | None]:
    """Map a server-side EndDevice href to the LFDI of the device it addresses.

    A control names a device by href; the connector registry is keyed by LFDI.
    The discovered EndDevice carries both, so the mapping comes from what the
    server said rather than from a guess. Returning one configured LFDI for
    every href would mean that with two or more devices every control resolves
    to the wrong connector, or to none, and is silently never applied.

    ``sole_device`` is the fallback for the single-device case, where the
    mapping is unambiguous and discovery may not have run yet. With several
    devices there is no safe fallback, so an unknown href resolves to nothing
    and the dispatcher reports it rather than guessing.
    """

    def resolve(device_href: str) -> str | None:
        client = get_client()
        state = client.state.end_devices.get(device_href) if client is not None else None
        if state is not None:
            lfdi = state.lfdi
            return (lfdi.hex() if isinstance(lfdi, bytes) else str(lfdi)).lower()
        return sole_device

    return resolve

connect_with_retry async

connect_with_retry(client: CsipClient, config: ClientConfig, stop: Event | None = None, state: ConnectionState | None = None, reconnect: Event | None = None) -> bool

Connect, retrying with backoff while the server is unreachable.

Returns True once connected, False if the attempt budget ran out or the caller asked to stop. A client on a gateway usually outlives the server's outages, so the default is to keep trying rather than exit and depend on something else noticing.

Only a transport failure is retried: a connection error, or the TLS error the transport raises once its own handshake retries are exhausted. The HTTP layer classifies that second one as an unreachable peer, and it is raised for a server-side handshake problem as readily as a local one, so it is retried rather than treated as fatal. Certificate material that is genuinely unusable -- missing, malformed, or a key that does not match -- fails when the SSL context is built, before any of this, and is reported as a configuration fault there.

A protocol or payload error is not retried. It produces the same result every time, so retrying only delays the operator seeing it -- and with the default of retrying forever, it would never be seen at all.

stop is awaited alongside the backoff, so a shutdown signal during an outage takes effect immediately. Without it, a systemd stop would hang until the server came back, which is exactly when it will not.

state is updated as the attempts proceed so the management API can report the outage, and reconnect short-circuits the wait when an operator asks for a retry now rather than at the end of the backoff.

Source code in src/py20305/cli.py

async def connect_with_retry(
    client: CsipClient,
    config: ClientConfig,
    stop: asyncio.Event | None = None,
    state: ConnectionState | None = None,
    reconnect: asyncio.Event | None = None,
) -> bool:
    """Connect, retrying with backoff while the server is unreachable.

    Returns True once connected, False if the attempt budget ran out or the
    caller asked to stop. A client on a gateway usually outlives the server's
    outages, so the default is to keep trying rather than exit and depend on
    something else noticing.

    Only a *transport* failure is retried: a connection error, or the TLS
    error the transport raises once its own handshake retries are exhausted.
    The HTTP layer classifies that second one as an unreachable peer, and it
    is raised for a server-side handshake problem as readily as a local one,
    so it is retried rather than treated as fatal. Certificate material that
    is genuinely unusable -- missing, malformed, or a key that does not match
    -- fails when the SSL context is built, before any of this, and is
    reported as a configuration fault there.

    A protocol or payload error is not retried. It produces the same result
    every time, so retrying only delays the operator seeing it -- and with the
    default of retrying forever, it would never be seen at all.

    ``stop`` is awaited alongside the backoff, so a shutdown signal during an
    outage takes effect immediately. Without it, a systemd stop would hang
    until the server came back, which is exactly when it will not.

    ``state`` is updated as the attempts proceed so the management API can
    report the outage, and ``reconnect`` short-circuits the wait when an
    operator asks for a retry now rather than at the end of the backoff.
    """
    delay = min(config.connection.initial_delay_seconds, config.connection.max_delay_seconds)
    attempt = 0

    while True:
        if stop is not None and stop.is_set():
            logger.info("stop requested before a connection was established")
            return False

        attempt += 1
        if state is not None:
            state.phase = "connecting"
            state.attempts = attempt
            state.retry_in_seconds = None
        try:
            await client.connect()
        except (Sep2ConnectionError, Sep2TlsError) as exc:
            budget = config.connection.max_attempts
            exhausted = budget and attempt >= budget
            if exhausted or not config.connection.retry_forever:
                logger.error("could not reach the server after %d attempt(s): %s", attempt, exc)
                if state is not None:
                    state.phase = "failed"
                    state.detail = str(exc)
                return False

            logger.warning(
                "connection attempt %d failed (%s); retrying in %.0fs", attempt, exc, delay
            )
            if state is not None:
                state.phase = "disconnected"
                state.detail = str(exc)
                state.retry_in_seconds = delay
            if await _sleep_or_stop(delay, stop, reconnect):
                if stop is not None and stop.is_set():
                    logger.info("stop requested while waiting to retry")
                    return False
                logger.info("reconnect requested; retrying now")
                if reconnect is not None:
                    reconnect.clear()
            delay = min(
                delay * config.connection.backoff_factor,
                config.connection.max_delay_seconds,
            )
        else:
            if attempt > 1:
                logger.info("connected after %d attempts", attempt)
            if state is not None:
                state.phase = "connected"
                state.detail = None
                state.retry_in_seconds = None
            return True

run async

run(config: ClientConfig) -> int

Run until stopped, returning the process exit code.

Source code in src/py20305/cli.py

async def run(config: ClientConfig) -> int:
    """Run until stopped, returning the process exit code."""
    stop = asyncio.Event()
    reconnect = asyncio.Event()
    connection = ConnectionState()
    # Installed before discovery, not after, so Ctrl-C during a retrying
    # discovery stops the process rather than being ignored until it finds
    # something -- which, on a network with no server, is never.
    _install_signal_handlers(stop)

    logger.info("%s", get_version_string())

    # Started before discovery, not after it. Announcing does not depend on
    # finding a server, and a client that only became visible once it had
    # found one would be invisible during exactly the local outage where
    # someone goes looking for it. It is also why this is torn down in the
    # `finally` below rather than alongside the client.
    advertiser = build_advertiser(config)
    if advertiser is not None:
        advertiser.start()

    try:
        return await _run_client(config, stop, reconnect, connection)
    finally:
        if advertiser is not None:
            # In a thread: stop() sends the goodbye records and joins.
            await asyncio.to_thread(advertiser.stop)

main

main(argv: list[str] | None = None) -> int

Entry point. Returns the process exit code.

Source code in src/py20305/cli.py

def main(argv: list[str] | None = None) -> int:
    """Entry point. Returns the process exit code."""
    args = parse_args(argv)

    try:
        config = load_config(args.config)
    except ConfigError as exc:
        # Before logging is configured, so this goes straight to stderr.
        print(f"error: {exc}", file=sys.stderr)
        return EXIT_CONFIG_ERROR

    if args.log_level:
        config = config.model_copy(
            update={"logging": config.logging.model_copy(update={"level": args.log_level})}
        )
    if args.multicast_transport:
        try:
            config = _apply_transport_override(config, args.multicast_transport)
        except ConfigError as exc:
            print(f"error: {exc}", file=sys.stderr)
            return EXIT_CONFIG_ERROR

    try:
        setup_logging(config.logging)
    except OSError as exc:
        # An unwritable log directory is a configuration fault like any other,
        # and it happens before logging exists to report it -- so this goes to
        # stderr and exits with the documented code rather than as a traceback.
        print(f"error: could not configure logging: {exc}", file=sys.stderr)
        return EXIT_CONFIG_ERROR

    if args.check:
        try:
            lfdi = compute_lfdi(config.tls.client_cert.read_text())
        except (OSError, ValueError) as exc:
            # ValueError as well as OSError: a certificate can be present and
            # unreadable *as a certificate*, which is a configuration problem
            # and should report as one rather than as a traceback.
            print(f"error: could not read the client certificate: {exc}", file=sys.stderr)
            return EXIT_CONFIG_ERROR
        server = config.server.url or (
            f"(discovered at startup over {config.discovery.transport})"
        )
        print(f"configuration OK: {args.config}")
        print(f"  server      : {server}")
        print(f"  client LFDI : {lfdi}")
        print(f"  client SFDI : {compute_sfdi(lfdi)}")
        print(f"  devices     : {len(config.devices)}")
        finding = f"on, {config.discovery.transport}" if config.discovery.enabled else "off"
        print(f"  discovery   : {finding}")
        print(f"  announcing  : {config.advertise.transport}")
        return EXIT_OK

    if args.discover:
        return _report_discovery(config)

    try:
        return asyncio.run(run(config))
    except (OSError, ValueError, ssl.SSLError) as exc:
        # The TLS material is only truly checked when it is loaded, which is
        # inside the run. A missing key, malformed PEM or a key that does not
        # match its certificate arrives here, and it is a configuration fault:
        # it deserves the documented exit code rather than a traceback and the
        # generic 1 that tells a supervisor nothing.
        #
        # Sep2TlsError is deliberately absent. The transport raises it once its
        # own handshake retries are exhausted, and the HTTP layer treats it as
        # an unreachable peer -- so a server-side handshake problem would stop
        # the process here with the documented systemd unit then refusing to
        # restart it. It is retried as a connection failure instead.
        logger.error("could not start: %s", exc)
        return EXIT_CONFIG_ERROR
    except KeyboardInterrupt:
        # Windows, where add_signal_handler is unavailable and Ctrl-C arrives
        # as an exception instead.
        return EXIT_OK