Telemetry

Posting measurements and DER resources back to the server.

py20305.telemetry.coordinator

Composition of the two telemetry managers for one running client.

TelemetryManager mirrors readings as MirrorUsagePoints; DerResourceManager PUTs DERCapability, DERSettings and DERStatus. Both need the same things -- which devices, at what rate, to which discovered hrefs -- and both have to be restarted when rediscovery moves those hrefs. That shared bookkeeping lives here rather than in the runner so a host application embedding this library starts telemetry the same way the runner does, instead of reimplementing it and drifting.

The coordinator owns lifecycle, not protocol: every decision about what a resource contains stays in the managers it composes.

TelemetryCoordinator

Starts, restarts and stops the telemetry managers for a set of devices.

Parameters:

Name	Type	Description	Default
client	CsipClient	The connected client. Its discovered state supplies every href, so setup belongs after discovery rather than before it.	required
lfdis	Iterable[str]	The devices to report on, as configured. Held as given: a device the server has temporarily forgotten stays registered, because configuration says what to manage and discovery says only where to send it.	required
connector_resolver	Callable[[str], BaseConnector | Awaitable[BaseConnector]]	Resolves an LFDI to its connector. Taken rather than derived from the client's dispatcher, so telemetry keeps no dependency on the connector layer and a host application can pass its own.	required
post_rate_seconds	int	Fallback interval for readings and DERStatus, used when the server's EndDevice.postRate does not say.	DEFAULT_POST_RATE
der_capability_poll_rate_seconds	int	Interval for DERCapability PUTs.	DEFAULT_CAPABILITY_POLL_RATE
der_settings_poll_rate_seconds	int	Interval for DERSettings PUTs.	DEFAULT_SETTINGS_POLL_RATE
device_telemetry	DeviceTelemetryEmitter | None	Reports each device read to a monitoring system.	None
source	MeasurementSource | None	Where measurements come from. Omitted, the metering cycle reads the connector directly; a host application serving several interfaces passes a store-backed source so a concurrent read does not become a second device poll.	None
is_provisioned	Callable[[str], bool] | None	Optional per-cycle gate on mirroring a device.	None
command_observer	CommandObserver | None	Offered the DERStatus the resource manager already reads, as confirmation evidence for outstanding commands.	None
reading_profile	ReadingProfile	How MirrorMeterReadings are encoded; see :class:~py20305.telemetry.mup.ReadingProfile.	DEFAULT

Source code in src/py20305/telemetry/coordinator.py

class TelemetryCoordinator:
    """Starts, restarts and stops the telemetry managers for a set of devices.

    Args:
        client: The connected client. Its discovered state supplies every
            href, so ``setup`` belongs after discovery rather than before it.
        lfdis: The devices to report on, as configured. Held as given: a
            device the server has temporarily forgotten stays registered,
            because configuration says what to manage and discovery says only
            where to send it.
        connector_resolver: Resolves an LFDI to its connector. Taken rather
            than derived from the client's dispatcher, so telemetry keeps no
            dependency on the connector layer and a host application can pass
            its own.
        post_rate_seconds: Fallback interval for readings and DERStatus, used
            when the server's ``EndDevice.postRate`` does not say.
        der_capability_poll_rate_seconds: Interval for DERCapability PUTs.
        der_settings_poll_rate_seconds: Interval for DERSettings PUTs.
        device_telemetry: Reports each device read to a monitoring system.
        source: Where measurements come from. Omitted, the metering cycle
            reads the connector directly; a host application serving several
            interfaces passes a store-backed source so a concurrent read does
            not become a second device poll.
        is_provisioned: Optional per-cycle gate on mirroring a device.
        command_observer: Offered the DERStatus the resource manager already
            reads, as confirmation evidence for outstanding commands.
        reading_profile: How MirrorMeterReadings are encoded; see
            :class:`~py20305.telemetry.mup.ReadingProfile`.
    """

    def __init__(
        self,
        client: CsipClient,
        *,
        lfdis: Iterable[str],
        connector_resolver: Callable[[str], BaseConnector | Awaitable[BaseConnector]],
        post_rate_seconds: int = DEFAULT_POST_RATE,
        der_capability_poll_rate_seconds: int = DEFAULT_CAPABILITY_POLL_RATE,
        der_settings_poll_rate_seconds: int = DEFAULT_SETTINGS_POLL_RATE,
        device_telemetry: DeviceTelemetryEmitter | None = None,
        source: MeasurementSource | None = None,
        is_provisioned: Callable[[str], bool] | None = None,
        command_observer: CommandObserver | None = None,
        reading_profile: ReadingProfile = ReadingProfile.DEFAULT,
    ) -> None:
        self._client = client
        self._reading_profile = reading_profile
        self._resolve_connector = connector_resolver
        self._post_rate = post_rate_seconds
        self._capability_poll_rate = der_capability_poll_rate_seconds
        self._settings_poll_rate = der_settings_poll_rate_seconds
        self._device_telemetry = device_telemetry
        self._source = source
        self._is_provisioned = is_provisioned
        self._command_observer = command_observer
        self._started: dict[str, bool] = {lfdi.lower(): False for lfdi in lfdis}
        self._telemetry: TelemetryManager | None = None
        self._der_resources: DerResourceManager | None = None
        #: Last MirrorUsagePointList path seen. Rediscovery clears state before
        #: repopulating it, and posting through that window has to go somewhere.
        #: Non-empty whenever the metering manager exists, because that manager
        #: is only built once a path has been discovered.
        self._last_mup_list_href: str = ""
        #: Whether the missing-MirrorUsagePointList warning has been said. Every
        #: rediscovery re-attempts the metering setup, so without this a server
        #: that never mirrors readings would log the same line on every
        #: structural change.
        self._warned_no_mup = False

    @property
    def telemetry(self) -> TelemetryManager | None:
        """The metering manager, or None when the server mirrors nothing."""
        return self._telemetry

    @property
    def der_resources(self) -> DerResourceManager | None:
        """The DER resource manager, once ``setup`` has run."""
        return self._der_resources

    def setup(self) -> None:
        """Construct both managers. Safe to call again; each part is idempotent."""
        self._setup_telemetry()
        self._setup_der_resources()

    def _setup_telemetry(self) -> None:
        """Build the metering manager, if the server offers somewhere to post.

        Re-attempted on rediscovery, so a server that brings the
        MirrorUsagePoint function set online after connect starts mirroring
        without a restart.
        """
        if self._telemetry is not None:
            return

        mup_list_href = self._client.state.mup_list_href
        if not mup_list_href:
            # A warning rather than an error: this is a server that does not
            # mirror readings, which is a valid deployment. Said once, because
            # a cycle that fails forever tells the operator less.
            if not self._warned_no_mup:
                logger.warning(
                    "the server exposed no MirrorUsagePointList; no readings will be posted. "
                    "DER resource PUTs are unaffected"
                )
                self._warned_no_mup = True
            return

        from py20305.telemetry.manager import TelemetryManager

        self._last_mup_list_href = mup_list_href
        self._warned_no_mup = False
        self._telemetry = TelemetryManager(
            client=self._client.http,
            # Read live: an upstream restart makes the client rediscover, and
            # the MirrorUsagePointList can move.
            mup_list_href=self._mup_list_href,
            connector_resolver=self._resolve_connector,
            is_provisioned=self._is_provisioned,
            source=self._source,
            device_telemetry=self._device_telemetry,
            reading_profile=self._reading_profile,
        )

    def _setup_der_resources(self) -> None:
        """Build the DER resource manager.

        Unconditional, unlike metering: the hrefs it PUTs to are per-device and
        resolved at start time, and a server that mirrors no readings still
        expects DERCapability.
        """
        if self._der_resources is not None:
            return

        from py20305.telemetry.der_resource_manager import DerResourceManager

        self._der_resources = DerResourceManager(
            client=self._client.http,
            connector_resolver=self._resolve_connector,
            command_observer=self._command_observer,
        )

    def _mup_list_href(self) -> str:
        current = self._client.state.mup_list_href
        if current:
            self._last_mup_list_href = current
        return current or self._last_mup_list_href

    def start_device_telemetry(self, lfdi: str | None = None) -> None:
        """Start metering and DER resource PUTs for one device, or all of them.

        Devices already started are left alone, so this is safe to call when a
        single device is added. Naming a device that was not in ``lfdis``
        registers it: a host application that registers devices while the
        client runs is making the request explicitly, and skipping it silently
        would report nothing for that device.

        Args:
            lfdi: One device, or None for every registered device.
        """
        if lfdi is not None:
            self._started.setdefault(lfdi.lower(), False)
        targets = [lfdi.lower()] if lfdi is not None else list(self._started)
        for target in targets:
            if self._started.get(target, True):
                continue
            post_rate = self._device_post_rate(target)
            if self._telemetry is not None:
                log_href, avail_href = self._telemetry_hrefs(target)
                logger.info(
                    "starting telemetry for %s: post_rate=%ds, log_events=%s, availability=%s",
                    target[:8],
                    post_rate,
                    log_href,
                    avail_href,
                )
                self._telemetry.start_metering(
                    target,
                    post_rate,
                    log_event_list_href=log_href,
                    der_availability_href=avail_href,
                )
            self._start_der_resources_for_device(target, post_rate)
            self._started[target] = True

    def stop_device_telemetry(self, lfdi: str) -> None:
        """Stop reporting one device and unregister it.

        For a device the server has deleted, or one an operator removed. Both
        managers drop it, and a later rediscovery does not bring it back --
        unlike a device that is merely absent from the current discovery, which
        stays registered.

        Args:
            lfdi: Device LFDI.
        """
        target = lfdi.lower()
        if self._telemetry is not None:
            self._telemetry.stop_metering(target)
        if self._der_resources is not None:
            self._der_resources.stop_device(target)
        self._started.pop(target, None)

    async def restart_device_telemetry(self) -> None:
        """Re-read every href from freshly discovered state.

        Wired to the client's structural-change hook: an upstream restart or a
        topology change moves the paths both managers PUT to, and a snapshot
        taken at first discovery would keep posting where the server no longer
        serves. Devices absent from the new state stay registered and are
        started with no hrefs, which makes their cycles no-ops until the server
        publishes them again.
        """
        # Late MirrorUsagePointList adoption. Cheap to attempt, and the only
        # path by which a server that added the function set after connect ever
        # gets readings.
        self._setup_telemetry()

        for target in self._started:
            self._started[target] = False
        self.start_device_telemetry()

    def _device_post_rate(self, lfdi: str) -> int:
        """The server's ``EndDevice.postRate``, falling back to configuration.

        IEEE 2030.5 gives the server the say over how often subordinate
        resources are posted. Zero and absent both mean it declined to specify.
        """
        for state in self._client.state.end_devices.values():
            if state.lfdi.hex().lower() != lfdi:
                continue
            server_rate = state.device.post_rate
            if server_rate is not None and server_rate > 0:
                return int(server_rate)
        return self._post_rate

    def _telemetry_hrefs(self, lfdi: str) -> tuple[str | None, str | None]:
        """LogEventList and DERAvailability hrefs for a device.

        Several EndDevices can carry one LFDI, so hrefs are merged across every
        match, first non-None winning.
        """
        log_href: str | None = None
        avail_href: str | None = None
        for state in self._client.state.end_devices.values():
            if state.lfdi.hex().lower() != lfdi:
                continue
            if log_href is None:
                log_href = state.log_event_list_href
            if avail_href is None:
                avail_href = state.der_availability_href
        return log_href, avail_href

    def _start_der_resources_for_device(self, lfdi: str, post_rate: int) -> None:
        """Schedule the three DER resource PUTs using discovered hrefs.

        DERStatus follows the posting rate, because it is the one of the three
        that tracks what the device is doing now. Capability and settings keep
        their own rates.
        """
        if self._der_resources is None:
            return

        capability_href: str | None = None
        settings_href: str | None = None
        status_href: str | None = None
        for state in self._client.state.end_devices.values():
            if state.lfdi.hex().lower() != lfdi:
                continue
            capability_href = capability_href or state.der_capability_href
            settings_href = settings_href or state.der_settings_href
            status_href = status_href or state.der_status_href

        self._der_resources.start_device(
            lfdi=lfdi,
            capability_href=capability_href,
            settings_href=settings_href,
            status_href=status_href,
            capability_poll_rate=self._capability_poll_rate,
            settings_poll_rate=self._settings_poll_rate,
            status_poll_rate=post_rate,
        )

    async def shutdown(self) -> None:
        """Stop metering, then the DER resource PUTs.

        Metering first because it is the noisier of the two: a reading posted
        during teardown is a reading the server has to reconcile, while a
        DERStatus PUT is idempotent.
        """
        if self._telemetry is not None:
            await self._telemetry.shutdown()
        if self._der_resources is not None:
            await self._der_resources.shutdown()

telemetry property

telemetry: TelemetryManager | None

The metering manager, or None when the server mirrors nothing.

der_resources property

der_resources: DerResourceManager | None

The DER resource manager, once setup has run.

setup

setup() -> None

Construct both managers. Safe to call again; each part is idempotent.

Source code in src/py20305/telemetry/coordinator.py

def setup(self) -> None:
    """Construct both managers. Safe to call again; each part is idempotent."""
    self._setup_telemetry()
    self._setup_der_resources()

start_device_telemetry

start_device_telemetry(lfdi: str | None = None) -> None

Start metering and DER resource PUTs for one device, or all of them.

Devices already started are left alone, so this is safe to call when a single device is added. Naming a device that was not in lfdis registers it: a host application that registers devices while the client runs is making the request explicitly, and skipping it silently would report nothing for that device.

Parameters:

Name	Type	Description	Default
lfdi	str | None	One device, or None for every registered device.	None

Source code in src/py20305/telemetry/coordinator.py

def start_device_telemetry(self, lfdi: str | None = None) -> None:
    """Start metering and DER resource PUTs for one device, or all of them.

    Devices already started are left alone, so this is safe to call when a
    single device is added. Naming a device that was not in ``lfdis``
    registers it: a host application that registers devices while the
    client runs is making the request explicitly, and skipping it silently
    would report nothing for that device.

    Args:
        lfdi: One device, or None for every registered device.
    """
    if lfdi is not None:
        self._started.setdefault(lfdi.lower(), False)
    targets = [lfdi.lower()] if lfdi is not None else list(self._started)
    for target in targets:
        if self._started.get(target, True):
            continue
        post_rate = self._device_post_rate(target)
        if self._telemetry is not None:
            log_href, avail_href = self._telemetry_hrefs(target)
            logger.info(
                "starting telemetry for %s: post_rate=%ds, log_events=%s, availability=%s",
                target[:8],
                post_rate,
                log_href,
                avail_href,
            )
            self._telemetry.start_metering(
                target,
                post_rate,
                log_event_list_href=log_href,
                der_availability_href=avail_href,
            )
        self._start_der_resources_for_device(target, post_rate)
        self._started[target] = True

stop_device_telemetry

stop_device_telemetry(lfdi: str) -> None

Stop reporting one device and unregister it.

For a device the server has deleted, or one an operator removed. Both managers drop it, and a later rediscovery does not bring it back -- unlike a device that is merely absent from the current discovery, which stays registered.

Parameters:

Name	Type	Description	Default
lfdi	str	Device LFDI.	required

Source code in src/py20305/telemetry/coordinator.py

def stop_device_telemetry(self, lfdi: str) -> None:
    """Stop reporting one device and unregister it.

    For a device the server has deleted, or one an operator removed. Both
    managers drop it, and a later rediscovery does not bring it back --
    unlike a device that is merely absent from the current discovery, which
    stays registered.

    Args:
        lfdi: Device LFDI.
    """
    target = lfdi.lower()
    if self._telemetry is not None:
        self._telemetry.stop_metering(target)
    if self._der_resources is not None:
        self._der_resources.stop_device(target)
    self._started.pop(target, None)

restart_device_telemetry async

restart_device_telemetry() -> None

Re-read every href from freshly discovered state.

Wired to the client's structural-change hook: an upstream restart or a topology change moves the paths both managers PUT to, and a snapshot taken at first discovery would keep posting where the server no longer serves. Devices absent from the new state stay registered and are started with no hrefs, which makes their cycles no-ops until the server publishes them again.

Source code in src/py20305/telemetry/coordinator.py

async def restart_device_telemetry(self) -> None:
    """Re-read every href from freshly discovered state.

    Wired to the client's structural-change hook: an upstream restart or a
    topology change moves the paths both managers PUT to, and a snapshot
    taken at first discovery would keep posting where the server no longer
    serves. Devices absent from the new state stay registered and are
    started with no hrefs, which makes their cycles no-ops until the server
    publishes them again.
    """
    # Late MirrorUsagePointList adoption. Cheap to attempt, and the only
    # path by which a server that added the function set after connect ever
    # gets readings.
    self._setup_telemetry()

    for target in self._started:
        self._started[target] = False
    self.start_device_telemetry()

shutdown async

shutdown() -> None

Stop metering, then the DER resource PUTs.

Metering first because it is the noisier of the two: a reading posted during teardown is a reading the server has to reconcile, while a DERStatus PUT is idempotent.

Source code in src/py20305/telemetry/coordinator.py

async def shutdown(self) -> None:
    """Stop metering, then the DER resource PUTs.

    Metering first because it is the noisier of the two: a reading posted
    during teardown is a reading the server has to reconcile, while a
    DERStatus PUT is idempotent.
    """
    if self._telemetry is not None:
        await self._telemetry.shutdown()
    if self._der_resources is not None:
        await self._der_resources.shutdown()

py20305.telemetry.manager

TelemetryManager for IEEE 2030.5 MirrorUsagePoint posting.

Manages the lifecycle of telemetry posting for DER devices: - Creates MUP on first metering cycle - Posts meter readings on subsequent cycles - Handles 404 recovery by re-creating MUP - Posts LogEvents when alarm status is non-zero - PUTs DERAvailability when href is discovered

DeviceTelemetryState dataclass

Per-device telemetry tracking state.

Source code in src/py20305/telemetry/manager.py

@dataclass
class DeviceTelemetryState:
    """Per-device telemetry tracking state."""

    lfdi: str
    mup_posted: bool = False
    mup_href: str | None = None
    #: mRID slots the last successful MUP POST registered. Kept across a 400/404
    #: reset so the re-created MUP still registers readings absent that cycle.
    registered_slots: frozenset[int] = frozenset()
    #: ACType the last successful MUP POST was built with. The totals' phase
    #: code derives from it, so a cycle without ACType falls back to it rather
    #: than re-POSTing with no phase, and a change to it re-POSTs.
    registered_ac_type: int | None = None
    #: Slot set and ACType of a re-POST the server refused with a 4xx. Not
    #: retried until the registration changes again or the process restarts;
    #: a 5xx or a connection error is retried every cycle.
    rejected_registration: tuple[frozenset[int], int | None] | None = None
    post_rate: int = 300
    #: postRate read back from the created MUP (B.17.1). It outranks the
    #: ``post_rate`` a later ``start_metering`` passes in, since the MUP is
    #: not re-POSTed and its rate is not read back again.
    server_post_rate: int | None = None
    log_event_list_href: str | None = None
    der_availability_href: str | None = None
    log_event_id_counter: int = 0
    #: Last alarm bitmap seen from the connector. LogEvents are posted on
    #: TRANSITION (CSIP s4.6.3: alarms and their return-to-normal messages are
    #: reported "as they occur"), so a persisting alarm isn't re-posted every
    #: metering cycle and a cleared bit still produces its RTN event.
    last_alarm_status: int = 0
    posted_log_events: deque[dict[str, Any]] = field(default_factory=lambda: deque(maxlen=20))
    # Set when the server rejects a MUP / readings POST with 403 Forbidden
    # (a conformant server enforces IEEE 2030.5-2023 §10.11.3 Rule e: only the MUP's
    # creator may POST to it). Retrying with the same LFDI cannot make the
    # request succeed -- something external must change (server admin clears
    # state, the operator restarts metering with a different identity, etc.).
    # While ``telemetry_blocked`` is set, the metering cycle skips POST
    # attempts to avoid quietly hammering an unaccepting server.
    #
    # Sticky-flag invariants:
    # * Idempotent ``TelemetryManager.start_metering(lfdi=...)`` calls (a
    #   no-op restart from config reload, etc.) PRESERVE the block. The
    #   carry-over in ``start_metering`` copies ``telemetry_blocked`` /
    #   ``blocked_reason`` from any existing state for this LFDI; otherwise
    #   a re-call would clear the block and resume hammering.
    # * Process restart clears the block (the in-memory state is gone).
    # * ``stop_metering(lfdi)`` followed by ``start_metering(lfdi, ...)``
    #   clears the block. ``stop_metering`` removes the device entry
    #   entirely, so the next ``start_metering`` finds no existing state
    #   to inherit from and defaults to ``telemetry_blocked=False``. This
    #   is the documented recovery path for an operator who has fixed the
    #   server-side identity issue.
    telemetry_blocked: bool = False
    blocked_reason: str | None = None

TelemetryManager

Manages telemetry posting for all devices.

The manager handles the full MUP lifecycle: 1. On first metering cycle, creates and POSTs the MUP 2. On subsequent cycles, POSTs meter reading updates 3. On 404 response, resets state to re-create MUP

Additionally, on each cycle (after the MUP/readings stage): 4. POSTs LogEvent if alarm status is non-zero 5. PUTs DERAvailability if href is discovered

Each posting stage is independently error-handled so failures don't cascade across posting types.

LFDI keys are normalized to lowercase to avoid case mismatch bugs.

Source code in src/py20305/telemetry/manager.py

class TelemetryManager:
    """Manages telemetry posting for all devices.

    The manager handles the full MUP lifecycle:
    1. On first metering cycle, creates and POSTs the MUP
    2. On subsequent cycles, POSTs meter reading updates
    3. On 404 response, resets state to re-create MUP

    Additionally, on each cycle (after the MUP/readings stage):
    4. POSTs LogEvent if alarm status is non-zero
    5. PUTs DERAvailability if href is discovered

    Each posting stage is independently error-handled so failures
    don't cascade across posting types.

    LFDI keys are normalized to lowercase to avoid case mismatch bugs.
    """

    def __init__(
        self,
        client: Sep2Client,
        mup_list_href: str | Callable[[], str],
        connector_resolver: Callable[[str], BaseConnector | Awaitable[BaseConnector]],
        is_provisioned: Callable[[str], bool] | None = None,
        source: MeasurementSource | None = None,
        device_telemetry: DeviceTelemetryEmitter | None = None,
        reading_profile: ReadingProfile = ReadingProfile.DEFAULT,
    ) -> None:
        """Initialize the TelemetryManager.

        Args:
            client: IEEE 2030.5 HTTP client
            mup_list_href: Server path for MirrorUsagePointList, or a callable
                returning it. Pass a callable when the path can change under
                you -- rediscovery after an upstream restart moves it, and a
                snapshot would keep posting where the server no longer serves.
            is_provisioned: Optional gate. When set, the metering cycle posts a
                device's MirrorUsagePoint/readings only if ``is_provisioned(lfdi)``
                returns True (e.g. the LFDI is in the server's EndDeviceList).
                Re-checked each cycle, so mirroring starts/stops as the device is
                provisioned/removed. ``None`` (default) mirrors every device.
            connector_resolver: Callback to resolve LFDI -> connector.
                May be either sync (``Callable[[str], BaseConnector]``)
                or async (``Callable[[str], Awaitable[BaseConnector]]``);
                the metering cycle awaits the result iff it's awaitable.
                Async resolvers are preferred so first-touch construction
                of a real Modbus connector can offload its scan +
                readiness retries to a worker thread instead of blocking
                the event loop (see ``_aresolve_connector``
                on the client).
            source: Where measurements come from. Omitted, the manager reads
                the connector directly on every cycle, which is correct for a
                deployment with one upstream interface. A host application serving
                several passes a store-backed source instead, so the metering
                cycle and a concurrent management-API request collapse onto one
                device poll -- the manager does not know which it was given.
            device_telemetry: Reports each device read to the monitoring
                system. Covers the LogEvent-cycle status and availability
                reads this manager issues itself, and — when no ``source`` is
                passed — the metering reads of the source constructed here; a
                caller passing its own ``source`` configures that source's
                reporting there instead. Optional and disabled by default.
            reading_profile: How MirrorMeterReadings are encoded, for every
                device. Fixed for the manager's lifetime: a change of profile
                changes units and sign on existing mRIDs, so it is a restart.
        """
        self._client = client
        self._reading_profile = reading_profile
        # Server timebase for telemetry timestamp defaults. isinstance guard
        # keeps AsyncMock clients in tests (whose .timebase is a Mock)
        # falling back to an identity timebase; production Sep2Client always
        # carries the real shared instance.
        tb = getattr(client, "timebase", None)
        self._timebase = tb if isinstance(tb, ServerTimebase) else ServerTimebase()
        #: Read through a callable when one is given, because an upstream
        #: restart makes the client rediscover its resource paths and the
        #: MirrorUsagePointList can move. A snapshot taken at construction
        #: would keep posting to the path the server no longer serves.
        self._mup_list_href_source = (
            mup_list_href if callable(mup_list_href) else (lambda: mup_list_href)
        )
        self._resolve_connector = connector_resolver
        self._is_provisioned = is_provisioned
        # Held for the status/availability reads this manager issues itself;
        # the metering-poll reads reach it only through the source built here,
        # and a caller supplying its own `source` wires it there.
        self._telemetry = device_telemetry
        self._source = source or DirectConnectorSource(
            connector_resolver, telemetry=device_telemetry
        )
        self._devices: dict[str, DeviceTelemetryState] = {}
        self._scheduler = PollScheduler()

    def start_metering(
        self,
        lfdi: str,
        post_rate: int,
        log_event_list_href: str | None = None,
        der_availability_href: str | None = None,
    ) -> None:
        """Start metering loop for a device.

        Args:
            lfdi: Device LFDI
            post_rate: Posting interval in seconds
            log_event_list_href: Server path for LogEventList POST
            der_availability_href: Server path for DERAvailability PUT
        """
        lfdi_norm = lfdi.lower()
        existing = self._devices.get(lfdi_norm)
        new_state = DeviceTelemetryState(
            lfdi=lfdi_norm,
            post_rate=post_rate,
            log_event_list_href=log_event_list_href,
            der_availability_href=der_availability_href,
        )
        if existing is not None:
            # Preserve MUP state from a previous metering session so we
            # don't re-POST a MirrorUsagePoint that the server already
            # knows about.
            if existing.mup_posted:
                new_state.mup_posted = existing.mup_posted
                new_state.mup_href = existing.mup_href
                new_state.registered_slots = existing.registered_slots
                new_state.registered_ac_type = existing.registered_ac_type
                new_state.rejected_registration = existing.rejected_registration
                if existing.server_post_rate is not None:
                    new_state.server_post_rate = existing.server_post_rate
                    new_state.post_rate = existing.server_post_rate
                new_state.log_event_id_counter = existing.log_event_id_counter
            # Alarm-transition state is independent of MUP registration, so it
            # carries over unconditionally: an idempotent start_metering (config
            # reload) must not re-announce alarms that are already active.
            new_state.last_alarm_status = existing.last_alarm_status
            # Preserve the Rule e block across an idempotent start_metering
            # call (e.g. config reload, a host application starting device telemetry
            # re-running for an already-started device). Without this carry-
            # over, a re-call of start_metering would silently clear the
            # block and resume hammering an unaccepting server. The
            # documented recovery path is stop_metering -> start_metering:
            # stop_metering removes the device entry entirely, so the
            # second start_metering finds no `existing` to inherit from
            # and the new state defaults to telemetry_blocked=False.
            new_state.telemetry_blocked = existing.telemetry_blocked
            new_state.blocked_reason = existing.blocked_reason
        self._devices[lfdi_norm] = new_state
        # The planner acquires on this cadence; the metering cycle reads what
        # it stored. Declared as post_rate so a default deployment sees the
        # same device poll rate it did before acquisition was decoupled.
        self._source.declare(lfdi_norm, float(new_state.post_rate))
        callback: Callable[[], Awaitable[None]] = partial(self._metering_cycle, lfdi_norm)
        self._scheduler.schedule(
            f"metering_{lfdi_norm}",
            new_state.post_rate,
            callback,
        )
        logger.info(
            "Started metering for device %s at %ds interval", lfdi_norm[:8], new_state.post_rate
        )

    def stop_metering(self, lfdi: str) -> None:
        """Stop metering for a device.

        Args:
            lfdi: Device LFDI
        """
        lfdi_norm = lfdi.lower()
        if lfdi_norm in self._devices:
            del self._devices[lfdi_norm]
            self._source.release(lfdi_norm)
            # Scheduler will handle task cleanup on next cancel_all

    async def _read(self, connector: Any, method: str, lfdi: str) -> dict[str, Any]:
        """Read one resource off a device, reporting it to the monitoring system.

        These reads run on every metering cycle and do not pass through the
        measurement source, so without this they are routine client-initiated
        southbound traffic a monitoring system never sees. Mirrors
        ``DerResourceManager._read``.

        Args:
            connector: The device connector to read from.
            method: The connector method to call, e.g. ``"fetch_status"``.
            lfdi: The device's LFDI, for attribution.

        Returns:
            Whatever the connector returned, unchanged.
        """
        values = await getattr(connector, method)()
        if self._telemetry is not None and isinstance(values, dict):
            self._telemetry.record_read(lfdi, values, connector=connector, lfdi=lfdi)
        return values  # type: ignore[no-any-return]

    async def _metering_cycle(self, lfdi: str) -> None:
        """Single metering cycle: MUP/readings, then LogEvent and DERAvailability.

        Each posting stage is independently try/excepted so failures in one
        don't prevent the others from executing.
        """
        state = self._devices.get(lfdi)
        if state is None:
            logger.warning("Metering cycle for unknown device %s", lfdi[:8])
            return

        # Server enforced Rule e and rejected this device's POST. The blocked
        # flag is sticky -- retrying with the same LFDI can't recover, so we
        # short-circuit here rather than hammering the server every cycle.
        # The diagnostics entry that set the flag is the operator-visible
        # signal; the per-cycle log here stays at debug to avoid log spam.
        if state.telemetry_blocked:
            logger.debug(
                "Telemetry cycle skipped for %s (blocked: %s)",
                state.lfdi[:8],
                state.blocked_reason or "unspecified",
            )
            return

        # only_mirror_discovered_devices gate: skip mirroring a device that the
        # server hasn't provisioned (not in its EndDeviceList). Re-checked each
        # cycle, so it starts automatically once the device is provisioned.
        if self._is_provisioned is not None and not self._is_provisioned(lfdi):
            logger.debug(
                "Metering cycle for %s skipped: device not in the server's "
                "EndDeviceList (only_mirror_discovered_devices)",
                state.lfdi[:8],
            )
            return

        logger.debug(
            "Metering cycle for %s: mup_posted=%s, mup_href=%s",
            state.lfdi[:8],
            state.mup_posted,
            state.mup_href,
        )

        connector_name = "unknown"
        try:
            # First-touch construction can block (Modbus scan + retries);
            # the async resolver path offloads that to a worker thread on
            # the first call and short-circuits on subsequent cycles
            #. Sync resolvers are still accepted for backward
            # compat with embedded callers that wire in a custom callback.
            # Still resolved here for the LogEvent and DERAvailability stages,
            # which read the connector directly.
            result = self._resolve_connector(lfdi)
            connector = await result if inspect.isawaitable(result) else result
            connector_name = type(connector).__name__
        except Exception as e:
            self._report_acquisition_failure(lfdi, connector_name, e)
            return

        # The source decides whether that costs a device read. Freshness comes
        # back already resolved against whatever demand it knows about, so the
        # cycle below reasons about quality without knowing the cadence.
        #
        # The source resolves the connector itself rather than being handed the
        # one resolved above: taking a pre-resolved connector would put the
        # southbound object back into an interface whose whole purpose is to
        # keep it out. That costs a second resolver call per cycle, which is a
        # cache hit for any memoizing resolver -- the documented requirement on
        # both this source and AcquisitionService.
        snapshot = await self._source.read(lfdi)

        if snapshot.last_success is None:
            # Never read successfully -- there is no value to post, not even a
            # stale one. Distinct from comm-lost with history, which does post.
            self._report_acquisition_failure(lfdi, connector_name, snapshot.error)
            return

        if snapshot.quality is not Quality.GOOD:
            # Keep posting. MUP.013/014 are replace-on-receipt, so
            # omitting leaves the pre-outage value live on the server with no
            # staleness marker -- the fabrication that flagging avoids.
            self._report_acquisition_failure(lfdi, connector_name, snapshot.error)

        monitoring, overrides = _monitoring_payload(snapshot)
        # The acquisition instant, carried into the server timebase rather
        # than read from it. A second now() call would reintroduce post time.
        acquired_at = int(snapshot.last_success + self._timebase.offset())
        stale = snapshot.quality is not Quality.GOOD
        monitoring = self._with_registered_ac_type(state, monitoring)

        # Stage 1: MUP creation or meter readings
        if not state.mup_posted:
            await self._post_mup(state, monitoring, overrides)
        else:
            if self._registration_grew(state, monitoring):
                # A reading first supplied after the MUP was posted needs its
                # ReadingType registered before readings for it are posted
                # (Rule h.2); without it the server answers 400 (Rule h.3).
                await self._post_mup(state, monitoring, overrides)
            # Post cadence, not acquisition cadence -- see
            # create_meter_reading_list. The planner may read sooner (a tighter
            # consumer) or much later (a device in backoff); the next POST is
            # the one thing we can promise. Taken after any re-POST above, whose
            # readback may have changed post_rate.
            next_update = int(self._timebase.now()) + state.post_rate
            await self._post_readings(
                state,
                monitoring,
                overrides,
                timestamp=acquired_at,
                next_update_time=next_update,
                stale=stale,
            )

        # Stage 2: LogEvent POST (alarm-driven)
        await self._post_log_event(state, connector)

        # Stage 3: DERAvailability PUT
        await self._put_der_availability(state, connector)

    def _report_acquisition_failure(
        self, lfdi: str, connector_name: str, error: Exception | None
    ) -> None:
        """Report a failed monitoring acquisition to diagnostics.

        Kept here rather than in the store layer so the operator-facing
        identity (source, dedup key, wording) stays with the consumer that
        cares about it -- the management API describes the same failure
        differently.
        """
        if error is None:
            return

        from py20305.diagnostics import report

        details = {"lfdi": lfdi, "connector": connector_name, "error": str(error)}
        if isinstance(error, ConnectorError | OSError):
            report(
                "warnings",
                f"Connector unreachable for {lfdi[:8]} ({connector_name}): {error}",
                source="telemetry",
                dedup_key=f"telemetry:{lfdi}:connector_unreachable",
                details=details,
            )
        else:
            report(
                "warnings",
                f"Failed to fetch monitoring for {lfdi[:8]} from {connector_name}: {error}",
                source="telemetry",
                dedup_key=f"telemetry:{lfdi}:fetch_monitoring",
                details=details,
                exc_info=error,
            )

    async def _post_mup(
        self,
        state: DeviceTelemetryState,
        monitoring_data: dict[str, Any],
        overrides: dict[str, ReadingOverride] | None = None,
    ) -> None:
        """POST MUP to server, track location and server-preferred postRate.

        The first POST of a device's MUP posts no readings that cycle; they
        begin on the next one. A re-POST under SIGNED_LOAD_CONVENTION, made
        when the registered set grows, is followed by that cycle's readings.

        IEEE B.17.1: The server MAY modify postRate to indicate its preferred
        posting rate. We read back the created MUP to honour the server's rate.
        """
        slots = registration_slots(monitoring_data, self._reading_profile, state.registered_slots)
        mup = create_mup(
            state.lfdi,
            monitoring_data,
            state.post_rate,
            overrides,
            profile=self._reading_profile,
            registered=state.registered_slots,
        )
        body = to_xml(mup, server_2018_compat=self._client.server_2018_compat)

        try:
            location = await self._client.post_bytes(self._mup_list_href_source(), body)
            state.mup_posted = True
            state.registered_slots = slots
            state.registered_ac_type = ac_type_of(monitoring_data)
            state.rejected_registration = None

            if location:
                state.mup_href = location
                logger.debug("MUP posted for %s, location: %s", state.lfdi[:8], location)
                await self._readback_post_rate(state, location)
            else:
                logger.debug("MUP posted for %s, no location header", state.lfdi[:8])
        except Sep2ProtocolError as e:
            if state.mup_posted and 400 <= e.status_code < 500:
                # A re-POST of a MUP this LFDI already created. The server
                # refused the new registration, not the device -- a 403 too,
                # which here is not the Rule e block below -- so the registered
                # readings keep posting and this registration is not retried.
                state.rejected_registration = (slots, ac_type_of(monitoring_data))
            elif e.status_code == 403:
                # IEEE 2030.5-2023 §10.11.3 Rule e + a.4: the body's mRID
                # already exists on the server and is owned by a different
                # LFDI. The server treats this as a duplicate-mRID re-POST
                # from a non-creator and refuses. Same blocking semantics
                # as the readings 403 -- retry can't help.
                from py20305.diagnostics import report

                state.telemetry_blocked = True
                state.blocked_reason = (
                    f"403 Forbidden on POST {self._mup_list_href_source()} -- the MUP "
                    "mRID already exists on the server under a different "
                    "creator LFDI (Rule e). Manual intervention required."
                )
                report(
                    "errors",
                    (
                        f"Telemetry blocked for {state.lfdi[:8]}: server returned "
                        f"403 Forbidden on initial POST {self._mup_list_href_source()}. "
                        "The MUP's mRID is already owned by a different LFDI on "
                        "the server (per IEEE 2030.5-2023 §10.11.3 Rule e). "
                        "Subsequent metering cycles will skip until restarted."
                    ),
                    source="telemetry",
                    dedup_key=f"telemetry:{state.lfdi}:rule_e_blocked",
                    details={
                        "lfdi": state.lfdi,
                        "op": "mup_post",
                        "mup_list_href": self._mup_list_href_source(),
                        "status_code": 403,
                        "error": str(e),
                    },
                )
                return
            from py20305.client.errors import compat_hint_suffix
            from py20305.diagnostics import report

            hint = compat_hint_suffix(e, self._client.server_2018_compat)
            report(
                "warnings",
                (
                    f"Failed to POST MUP for {state.lfdi[:8]}: "
                    f"status={e.status_code}, error={e}{hint}"
                ),
                source="telemetry",
                dedup_key=f"telemetry:{state.lfdi}:mup_post",
                details={
                    "lfdi": state.lfdi,
                    "op": "mup_post",
                    "status_code": e.status_code,
                    "error": str(e),
                },
            )
        except Sep2ConnectionError as e:
            from py20305.diagnostics import report

            report(
                "warnings",
                f"Server unreachable posting MUP for {state.lfdi[:8]}: {e}",
                source="telemetry",
                dedup_key=f"telemetry:{state.lfdi}:mup_post",
                details={"lfdi": state.lfdi, "op": "mup_post", "error": str(e)},
            )
        except Exception as e:
            from py20305.diagnostics import report

            report(
                "warnings",
                f"Unexpected error posting MUP for {state.lfdi[:8]}: {e}",
                source="telemetry",
                dedup_key=f"telemetry:{state.lfdi}:mup_post",
                details={"lfdi": state.lfdi, "op": "mup_post", "error": str(e)},
                exc_info=True,
            )

    def _registration_grew(
        self, state: DeviceTelemetryState, monitoring_data: dict[str, Any]
    ) -> bool:
        """Whether the posted MUP no longer describes this cycle's readings.

        True when the cycle supplies a reading the MUP did not register, or
        reports a different ACType, which changes the totals' phase code. Only
        SIGNED_LOAD_CONVENTION registers readings as they appear. DEFAULT
        registers its set once, as it always has.
        """
        if self._reading_profile is not ReadingProfile.SIGNED_LOAD_CONVENTION:
            return False
        ac_type = ac_type_of(monitoring_data)
        slots = registration_slots(monitoring_data, self._reading_profile, state.registered_slots)
        if (slots, ac_type) == state.rejected_registration:
            return False
        return ac_type != state.registered_ac_type or slots != state.registered_slots

    def _with_registered_ac_type(
        self, state: DeviceTelemetryState, monitoring_data: dict[str, Any]
    ) -> dict[str, Any]:
        """Fill a missing ACType from the one the MUP was registered with.

        Under SIGNED_LOAD_CONVENTION the totals' phase code and the per-line
        block both derive from ACType, so a cycle that omits it must not
        re-POST the MUP as if the device had none.
        """
        if (
            self._reading_profile is not ReadingProfile.SIGNED_LOAD_CONVENTION
            or state.registered_ac_type is None
            or ac_type_of(monitoring_data) is not None
        ):
            return monitoring_data
        return {**monitoring_data, "ACType": state.registered_ac_type}

    async def _readback_post_rate(
        self,
        state: DeviceTelemetryState,
        mup_href: str,
    ) -> None:
        """GET the created MUP and adopt server-preferred postRate if changed."""
        try:
            mup = await self._client.get(mup_href, MirrorUsagePoint)
        except Exception:
            logger.warning(
                "Could not read back MUP postRate for %s; posting every %ds",
                state.lfdi[:8],
                state.post_rate,
            )
            return
        if mup.post_rate is None or mup.post_rate <= 0:
            return
        # The adopted rate also sets how often the device is read, so a server
        # cannot push it below MIN_POST_RATE: a device behind a slow serial link
        # is the thing least able to absorb a server asking for one post a
        # second. The floor never rises above the rate this client posted with,
        # so a server echoing a client's own short rate is agreeing with it.
        floor = min(MIN_POST_RATE, state.post_rate)
        rate = max(floor, mup.post_rate)
        if rate != mup.post_rate:
            logger.warning(
                "Server postRate for %s is %d s; posting every %d s, the minimum",
                state.lfdi[:8],
                mup.post_rate,
                rate,
            )
        state.server_post_rate = rate
        if rate == state.post_rate:
            return
        logger.info(
            "Server adjusted postRate for %s: %d -> %d",
            state.lfdi[:8],
            state.post_rate,
            rate,
        )
        state.post_rate = rate
        # Readings are both acquired and posted on this cadence; changing
        # only state.post_rate would leave both running at the old rate.
        self._source.declare(state.lfdi, float(rate))
        self._scheduler.set_interval(f"metering_{state.lfdi}", rate)

    async def _post_readings(
        self,
        state: DeviceTelemetryState,
        monitoring_data: dict[str, Any],
        overrides: dict[str, ReadingOverride] | None = None,
        *,
        timestamp: int | None = None,
        next_update_time: int | None = None,
        stale: bool = False,
    ) -> None:
        """POST meter readings to MUP location.

        ``timestamp`` is the acquisition instant in the server timebase.
        Defaults to post time only for callers that have no acquisition to
        cite, which after the poll planner is just tests.
        """
        # Discover MUP href if not known
        if not state.mup_href:
            state.mup_href = await self._discover_mup_href(state.lfdi)

        if not state.mup_href:
            logger.warning("No MUP href for %s, cannot post readings", state.lfdi[:8])
            return

        readings = create_meter_reading_list(
            state.lfdi,
            monitoring_data,
            timestamp=int(self._timebase.now()) if timestamp is None else timestamp,
            overrides=overrides,
            post_rate=state.post_rate,
            next_update_time=next_update_time,
            stale=stale,
            profile=self._reading_profile,
            registered=state.registered_slots,
        )
        body = to_xml(readings, server_2018_compat=self._client.server_2018_compat)

        try:
            await self._client.post_bytes(state.mup_href, body)
            logger.debug("Readings posted for %s", state.lfdi[:8])
        except Sep2ProtocolError as e:
            if e.status_code in (400, 404):
                # 404: MUP was deleted by server, reset state to recreate.
                # 400: server rejected the readings POST. The new IEEE
                # 2030.5-2023 §10.11.3 Rule h.3 enforcement on /mup/{N}
                # produces 400 in two scenarios we can recover from:
                # (a) version skew -- server upgraded ahead of a client
                # that still uses unstable mRIDs; (b) server lost
                # /upt/{N}/mr state while /mup/{N} survived. Both recover
                # via re-POST /mup, which goes through Rule a.4 and rebuilds
                # /upt/{N}/mr from the readings loop. We don't know for
                # sure that 400 means "stale state" (it could also be a
                # malformed body) but the recovery cost is low: a redundant
                # MUP re-POST either succeeds or also 400s, in which case
                # the operator sees a persistent failure on the MUP path
                # -- a much clearer signal than an infinite readings retry.
                logger.info("MUP %s for %s, will recreate", e.status_code, state.lfdi[:8])
                state.mup_posted = False
                state.mup_href = None
            elif e.status_code == 403:
                # A conformant server (IEEE 2030.5-2023 §10.11.3 Rule e) rejects
                # POSTs to a MUP from any LFDI other than its creator's.
                # Retrying with the same LFDI cannot recover. Mark the
                # device as blocked so subsequent cycles skip cleanly,
                # and surface to the operator at error level. Recreating
                # the MUP from a different LFDI is intentionally NOT
                # automatic -- it would either succeed (defeating the
                # protection) or also 403 with no progress.
                from py20305.diagnostics import report

                state.telemetry_blocked = True
                state.blocked_reason = (
                    f"403 Forbidden on POST {state.mup_href} -- the server "
                    "treats this LFDI as not the MUP creator (Rule e). "
                    "Manual intervention required."
                )
                report(
                    "errors",
                    (
                        f"Telemetry blocked for {state.lfdi[:8]}: server returned "
                        f"403 Forbidden on POST {state.mup_href}. The MUP was "
                        "created by a different client (per IEEE 2030.5-2023 "
                        "§10.11.3 Rule e). Subsequent metering cycles will "
                        "skip POSTs until the device is restarted or the "
                        "server-side state is cleared."
                    ),
                    source="telemetry",
                    dedup_key=f"telemetry:{state.lfdi}:rule_e_blocked",
                    details={
                        "lfdi": state.lfdi,
                        "op": "readings_post",
                        "mup_href": state.mup_href,
                        "status_code": 403,
                        "error": str(e),
                    },
                )
            else:
                from py20305.client.errors import compat_hint_suffix
                from py20305.diagnostics import report

                hint = compat_hint_suffix(e, self._client.server_2018_compat)
                report(
                    "warnings",
                    f"Failed to POST readings for {state.lfdi[:8]}: {e}{hint}",
                    source="telemetry",
                    dedup_key=f"telemetry:{state.lfdi}:readings_post",
                    details={
                        "lfdi": state.lfdi,
                        "op": "readings_post",
                        "status_code": e.status_code,
                        "error": str(e),
                    },
                )
        except Sep2ConnectionError as e:
            from py20305.diagnostics import report

            report(
                "warnings",
                f"Server unreachable posting readings for {state.lfdi[:8]}: {e}",
                source="telemetry",
                dedup_key=f"telemetry:{state.lfdi}:readings_post",
                details={"lfdi": state.lfdi, "op": "readings_post", "error": str(e)},
            )
        except Exception as e:
            from py20305.diagnostics import report

            report(
                "warnings",
                f"Unexpected error posting readings for {state.lfdi[:8]}: {e}",
                source="telemetry",
                dedup_key=f"telemetry:{state.lfdi}:readings_post",
                details={"lfdi": state.lfdi, "op": "readings_post", "error": str(e)},
                exc_info=True,
            )

    async def _post_log_event(
        self,
        state: DeviceTelemetryState,
        connector: BaseConnector,
    ) -> None:
        """POST LogEvents for DER alarm transitions (CSIP s4.6.3 / s5.2.5.3).

        One LogEvent per changed alarm bit: a newly-set bit posts its alarm
        code, a newly-cleared bit posts its return-to-normal code. A persisting
        alarm posts nothing (alarms are reported "as they occur"), and a clear
        to zero still posts the RTNs -- so this deliberately has no
        ``alarm == 0`` early-out.
        """
        if not state.log_event_list_href:
            return

        try:
            status = await self._read(connector, "fetch_status", state.lfdi)
            alarm = extract_alarm_status(status)
            events = alarm_bits_to_log_events(state.last_alarm_status, alarm)
            # Sync bits that can never produce a LogEvent (IEEE-reserved, 11+)
            # straight into the baseline -- there is nothing to post for them,
            # so they must not keep re-appearing in the diff every cycle. Bits
            # that DO map are advanced one at a time below, only once their
            # POST has landed.
            state.last_alarm_status = (state.last_alarm_status & MAPPED_ALARM_MASK) | (
                alarm & ~MAPPED_ALARM_MASK
            )

            unmapped = unmapped_alarm_bits(alarm)
            if unmapped:
                from py20305.diagnostics import report

                report(
                    "info",
                    f"DER alarm bits {unmapped} on {state.lfdi[:8]} have no "
                    "IEEE 2030.5-assigned LogEvent code (bits 11+ are reserved); "
                    "they are still reported in DERStatus alarmStatus.",
                    source="telemetry",
                    dedup_key=f"telemetry:{state.lfdi}:unmapped_alarm_bits",
                    details={"lfdi": state.lfdi, "bits": unmapped, "alarm_status": alarm},
                )

            for bit, code in events:
                await self._post_and_record_log_event(state, code, details=DER_ALARM_NAMES.get(bit))
                # Advance the baseline one bit at a time, AFTER the POST lands.
                # If a later POST fails, the bits already delivered stay synced
                # and the rest remain pending, so the next cycle retries exactly
                # the transitions the server never received (CSIP s4.6.3 requires
                # alarms and their RTNs to be reported).
                if alarm & (1 << bit):
                    state.last_alarm_status |= 1 << bit
                else:
                    state.last_alarm_status &= ~(1 << bit)
                logger.debug(
                    "LogEvent posted for %s: bit=%d code=%d (%s)",
                    state.lfdi[:8],
                    bit,
                    code,
                    DER_ALARM_NAMES.get(bit, "?"),
                )
        except (Sep2ConnectionError, ConnectorError, OSError) as e:
            from py20305.diagnostics import report

            report(
                "warnings",
                f"Failed to post LogEvent for {state.lfdi[:8]}: {e}",
                source="telemetry",
                dedup_key=f"telemetry:{state.lfdi}:logevent_post",
                details={"lfdi": state.lfdi, "op": "logevent_post", "error": str(e)},
            )
        except Exception as e:
            from py20305.diagnostics import report

            report(
                "warnings",
                f"Unexpected error posting LogEvent for {state.lfdi[:8]}: {e}",
                source="telemetry",
                dedup_key=f"telemetry:{state.lfdi}:logevent_post",
                details={"lfdi": state.lfdi, "op": "logevent_post", "error": str(e)},
                exc_info=True,
            )

    async def post_log_event(
        self,
        state: DeviceTelemetryState,
        alarm_status: int,
        details: str | None = None,
    ) -> None:
        """POST a manually triggered LogEvent for a device.

        Args:
            state: Device telemetry state (must have log_event_list_href).
            alarm_status: Alarm status bitmap.
            details: Optional human-readable detail string.

        Raises:
            Sep2ProtocolError: If the POST fails.
            ValueError: If device has no log_event_list_href.
        """
        if not state.log_event_list_href:
            raise ValueError(f"Device {state.lfdi[:8]} has no log_event_list_href")
        # Fan out one LogEvent per set alarm bit, using the IEEE-assigned code
        # for each. alarm_status=0 means "no alarms" and posts nothing -- code 0
        # is Over Current, so emitting it for an empty bitmap would report a
        # fault that isn't there.
        for bit, code in alarm_bits_to_log_events(0, alarm_status):
            await self._post_and_record_log_event(
                state, code, details=details or DER_ALARM_NAMES.get(bit)
            )
            logger.debug("LogEvent triggered for %s: bit=%d code=%d", state.lfdi[:8], bit, code)

    async def _post_and_record_log_event(
        self,
        state: DeviceTelemetryState,
        log_event_code: int,
        details: str | None = None,
    ) -> None:
        """POST one LogEvent and record it for the management API.

        Single place that builds the wire body and the ``posted_log_events``
        entry, so the code surfaced by ``GET /api/v1/logevents`` can't drift
        from the code actually sent.
        """
        assert state.log_event_list_href is not None  # callers check
        xml = create_log_event_xml(
            log_event_code,
            log_event_id=state.log_event_id_counter,
            details=details,
            server_2018_compat=self._client.server_2018_compat,
            created_time=int(self._timebase.now()),
        )
        await self._client.post_bytes(state.log_event_list_href, xml)
        state.posted_log_events.append(
            {
                "logEventID": state.log_event_id_counter,
                "functionSet": FUNCTION_SET_DER,
                "logEventCode": log_event_code,
                "logEventPEN": IEEE_2030_5_PEN,
                "profileID": PROFILE_IEEE_2030_5,
                "timestamp": time.time(),
            }
        )
        state.log_event_id_counter += 1

    _LOG_EVENT_BURST_COUNT = 5  # BASIC-027 requires 5 LogEvents per trigger

    async def post_log_event_burst(
        self,
        lfdi: str,
        alarm_status: int = 1,
        details: str | None = None,
        interval: float = 1.0,
    ) -> int:
        """Post a burst of LogEvents for BASIC-027 conformance.

        Posts exactly ``_LOG_EVENT_BURST_COUNT`` LogEvents with incrementing
        logEventIDs, spaced by *interval* seconds so each has a distinct
        createdDateTime.

        ``alarm_status`` must resolve to exactly ONE mapped alarm bit. The
        burst posts that bit's code N times, so the returned count is a true
        POST count (the caller derives ``log_event_ids`` from it). Routing a
        multi-bit bitmap through here would post N x bits events while
        reporting N -- and BASIC-027 wants exactly N.

        Args:
            lfdi: Device LFDI (case-insensitive).
            alarm_status: Alarm bitmap with exactly one IEEE-mapped bit set.
            details: Optional human-readable detail string.
            interval: Seconds between each POST (0 = no delay).

        Returns:
            Number of events successfully posted.

        Raises:
            ValueError: If the device is unknown, has no log_event_list_href,
                or ``alarm_status`` does not resolve to exactly one mapped bit.
        """
        state = self.get_device_state(lfdi)
        if state is None:
            raise ValueError(f"Device {lfdi[:8]}... not found in telemetry manager")
        if not state.log_event_list_href:
            raise ValueError(f"Device {lfdi[:8]}... has no log_event_list_href")

        events = alarm_bits_to_log_events(0, alarm_status)
        if len(events) != 1:
            raise ValueError(
                f"LogEvent burst needs exactly one IEEE-mapped alarm bit; "
                f"alarm_status=0x{alarm_status:X} resolves to {len(events)} "
                f"(0 = no mapped bit set -- bits 11+ are reserved; "
                f">1 = multi-bit, which would post {len(events)} events per iteration)"
            )
        bit, code = events[0]

        posted = 0
        for i in range(self._LOG_EVENT_BURST_COUNT):
            if interval > 0 and i > 0:
                await asyncio.sleep(interval)
            try:
                await self._post_and_record_log_event(
                    state, code, details=details or DER_ALARM_NAMES.get(bit)
                )
                posted += 1
            except Exception:
                logger.exception("LogEvent burst POST %d failed for %s", i, lfdi[:8])
                break

        logger.info(
            "LogEvent burst for %s: posted %d/%d events",
            lfdi[:8],
            posted,
            self._LOG_EVENT_BURST_COUNT,
        )
        return posted

    def find_device_with_log_events(self) -> str | None:
        """Return the first active device LFDI that has a log_event_list_href."""
        for lfdi, state in self._devices.items():
            if state.log_event_list_href:
                return lfdi
        return None

    async def _put_der_availability(
        self,
        state: DeviceTelemetryState,
        connector: BaseConnector,
    ) -> None:
        """PUT DERAvailability if href is discovered."""
        if not state.der_availability_href:
            return

        try:
            data = await self._read(connector, "fetch_availability", state.lfdi)
            model = build_der_availability(data, default_time=int(self._timebase.now()))
            await self._client.put_bytes(
                state.der_availability_href,
                to_xml(model, server_2018_compat=self._client.server_2018_compat),
            )
            logger.debug("DERAvailability PUT for %s", state.lfdi[:8])
        except (Sep2ConnectionError, ConnectorError, OSError) as e:
            from py20305.diagnostics import report

            report(
                "warnings",
                f"Failed to PUT DERAvailability for {state.lfdi[:8]}: {e}",
                source="telemetry",
                dedup_key=f"telemetry:{state.lfdi}:der_availability_put",
                details={
                    "lfdi": state.lfdi,
                    "op": "der_availability_put",
                    "error": str(e),
                },
            )
        except Exception as e:
            from py20305.diagnostics import report

            report(
                "warnings",
                f"Unexpected error putting DERAvailability for {state.lfdi[:8]}: {e}",
                source="telemetry",
                dedup_key=f"telemetry:{state.lfdi}:der_availability_put",
                details={
                    "lfdi": state.lfdi,
                    "op": "der_availability_put",
                    "error": str(e),
                },
                exc_info=True,
            )

    async def _discover_mup_href(self, lfdi: str) -> str | None:
        """Scan server MUP list to find our MUP by deviceLFDI.

        This is a fallback when the server didn't return a Location header
        after the MUP POST.
        """
        try:
            pages = await self._client.get_list(self._mup_list_href_source(), MirrorUsagePointList)
            for page in pages:
                for mup in page.mirror_usage_point or []:
                    if mup.device_lfdi and mup.device_lfdi.hex().lower() == lfdi:
                        logger.debug("Discovered MUP href for %s: %s", lfdi[:8], mup.href)
                        return mup.href
        except Sep2ConnectionError as e:
            from py20305.diagnostics import report

            report(
                "warnings",
                f"Server unreachable discovering MUP for {lfdi[:8]}: {e}",
                source="telemetry",
                dedup_key=f"telemetry:{lfdi}:mup_discovery",
                details={"lfdi": lfdi, "op": "mup_discovery", "error": str(e)},
            )
        except Exception as e:
            from py20305.diagnostics import report

            report(
                "warnings",
                f"Failed to discover MUP href for {lfdi[:8]}: {e}",
                source="telemetry",
                dedup_key=f"telemetry:{lfdi}:mup_discovery",
                details={"lfdi": lfdi, "op": "mup_discovery", "error": str(e)},
                exc_info=True,
            )

        return None

    @property
    def active_devices(self) -> list[str]:
        """Return list of LFDIs with active metering."""
        return list(self._devices.keys())

    def get_device_state(self, lfdi: str) -> DeviceTelemetryState | None:
        """Get telemetry state for a device."""
        return self._devices.get(lfdi.lower())

    def get_all_posted_log_events(self) -> list[dict[str, Any]]:
        """Return all posted log events across all devices."""
        events: list[dict[str, Any]] = []
        for state in self._devices.values():
            for ev in state.posted_log_events:
                events.append({**ev, "device": state.lfdi[:16]})
        events.sort(key=lambda e: e["timestamp"])
        return events

    async def shutdown(self) -> None:
        """Stop all metering tasks."""
        await self._scheduler.cancel_all()
        self._devices.clear()
        logger.info("TelemetryManager shutdown complete")

active_devices property

active_devices: list[str]

Return list of LFDIs with active metering.

start_metering

start_metering(lfdi: str, post_rate: int, log_event_list_href: str | None = None, der_availability_href: str | None = None) -> None

Start metering loop for a device.

Parameters:

Name	Type	Description	Default
lfdi	str	Device LFDI	required
post_rate	int	Posting interval in seconds	required
log_event_list_href	str | None	Server path for LogEventList POST	None
der_availability_href	str | None	Server path for DERAvailability PUT	None

Source code in src/py20305/telemetry/manager.py

def start_metering(
    self,
    lfdi: str,
    post_rate: int,
    log_event_list_href: str | None = None,
    der_availability_href: str | None = None,
) -> None:
    """Start metering loop for a device.

    Args:
        lfdi: Device LFDI
        post_rate: Posting interval in seconds
        log_event_list_href: Server path for LogEventList POST
        der_availability_href: Server path for DERAvailability PUT
    """
    lfdi_norm = lfdi.lower()
    existing = self._devices.get(lfdi_norm)
    new_state = DeviceTelemetryState(
        lfdi=lfdi_norm,
        post_rate=post_rate,
        log_event_list_href=log_event_list_href,
        der_availability_href=der_availability_href,
    )
    if existing is not None:
        # Preserve MUP state from a previous metering session so we
        # don't re-POST a MirrorUsagePoint that the server already
        # knows about.
        if existing.mup_posted:
            new_state.mup_posted = existing.mup_posted
            new_state.mup_href = existing.mup_href
            new_state.registered_slots = existing.registered_slots
            new_state.registered_ac_type = existing.registered_ac_type
            new_state.rejected_registration = existing.rejected_registration
            if existing.server_post_rate is not None:
                new_state.server_post_rate = existing.server_post_rate
                new_state.post_rate = existing.server_post_rate
            new_state.log_event_id_counter = existing.log_event_id_counter
        # Alarm-transition state is independent of MUP registration, so it
        # carries over unconditionally: an idempotent start_metering (config
        # reload) must not re-announce alarms that are already active.
        new_state.last_alarm_status = existing.last_alarm_status
        # Preserve the Rule e block across an idempotent start_metering
        # call (e.g. config reload, a host application starting device telemetry
        # re-running for an already-started device). Without this carry-
        # over, a re-call of start_metering would silently clear the
        # block and resume hammering an unaccepting server. The
        # documented recovery path is stop_metering -> start_metering:
        # stop_metering removes the device entry entirely, so the
        # second start_metering finds no `existing` to inherit from
        # and the new state defaults to telemetry_blocked=False.
        new_state.telemetry_blocked = existing.telemetry_blocked
        new_state.blocked_reason = existing.blocked_reason
    self._devices[lfdi_norm] = new_state
    # The planner acquires on this cadence; the metering cycle reads what
    # it stored. Declared as post_rate so a default deployment sees the
    # same device poll rate it did before acquisition was decoupled.
    self._source.declare(lfdi_norm, float(new_state.post_rate))
    callback: Callable[[], Awaitable[None]] = partial(self._metering_cycle, lfdi_norm)
    self._scheduler.schedule(
        f"metering_{lfdi_norm}",
        new_state.post_rate,
        callback,
    )
    logger.info(
        "Started metering for device %s at %ds interval", lfdi_norm[:8], new_state.post_rate
    )

stop_metering

stop_metering(lfdi: str) -> None

Stop metering for a device.

Parameters:

Name	Type	Description	Default
lfdi	str	Device LFDI	required

Source code in src/py20305/telemetry/manager.py

def stop_metering(self, lfdi: str) -> None:
    """Stop metering for a device.

    Args:
        lfdi: Device LFDI
    """
    lfdi_norm = lfdi.lower()
    if lfdi_norm in self._devices:
        del self._devices[lfdi_norm]
        self._source.release(lfdi_norm)

post_log_event async

post_log_event(state: DeviceTelemetryState, alarm_status: int, details: str | None = None) -> None

POST a manually triggered LogEvent for a device.

Parameters:

Name	Type	Description	Default
state	DeviceTelemetryState	Device telemetry state (must have log_event_list_href).	required
alarm_status	int	Alarm status bitmap.	required
details	str | None	Optional human-readable detail string.	None

Raises:

Type	Description
Sep2ProtocolError	If the POST fails.
ValueError	If device has no log_event_list_href.

Source code in src/py20305/telemetry/manager.py

async def post_log_event(
    self,
    state: DeviceTelemetryState,
    alarm_status: int,
    details: str | None = None,
) -> None:
    """POST a manually triggered LogEvent for a device.

    Args:
        state: Device telemetry state (must have log_event_list_href).
        alarm_status: Alarm status bitmap.
        details: Optional human-readable detail string.

    Raises:
        Sep2ProtocolError: If the POST fails.
        ValueError: If device has no log_event_list_href.
    """
    if not state.log_event_list_href:
        raise ValueError(f"Device {state.lfdi[:8]} has no log_event_list_href")
    # Fan out one LogEvent per set alarm bit, using the IEEE-assigned code
    # for each. alarm_status=0 means "no alarms" and posts nothing -- code 0
    # is Over Current, so emitting it for an empty bitmap would report a
    # fault that isn't there.
    for bit, code in alarm_bits_to_log_events(0, alarm_status):
        await self._post_and_record_log_event(
            state, code, details=details or DER_ALARM_NAMES.get(bit)
        )
        logger.debug("LogEvent triggered for %s: bit=%d code=%d", state.lfdi[:8], bit, code)

post_log_event_burst async

post_log_event_burst(lfdi: str, alarm_status: int = 1, details: str | None = None, interval: float = 1.0) -> int

Post a burst of LogEvents for BASIC-027 conformance.

Posts exactly _LOG_EVENT_BURST_COUNT LogEvents with incrementing logEventIDs, spaced by interval seconds so each has a distinct createdDateTime.

alarm_status must resolve to exactly ONE mapped alarm bit. The burst posts that bit's code N times, so the returned count is a true POST count (the caller derives log_event_ids from it). Routing a multi-bit bitmap through here would post N x bits events while reporting N -- and BASIC-027 wants exactly N.

Parameters:

Name	Type	Description	Default
lfdi	str	Device LFDI (case-insensitive).	required
alarm_status	int	Alarm bitmap with exactly one IEEE-mapped bit set.	1
details	str | None	Optional human-readable detail string.	None
interval	float	Seconds between each POST (0 = no delay).	1.0

Returns:

Type	Description
int	Number of events successfully posted.

Raises:

Type	Description
ValueError	If the device is unknown, has no log_event_list_href, or alarm_status does not resolve to exactly one mapped bit.

Source code in src/py20305/telemetry/manager.py

async def post_log_event_burst(
    self,
    lfdi: str,
    alarm_status: int = 1,
    details: str | None = None,
    interval: float = 1.0,
) -> int:
    """Post a burst of LogEvents for BASIC-027 conformance.

    Posts exactly ``_LOG_EVENT_BURST_COUNT`` LogEvents with incrementing
    logEventIDs, spaced by *interval* seconds so each has a distinct
    createdDateTime.

    ``alarm_status`` must resolve to exactly ONE mapped alarm bit. The
    burst posts that bit's code N times, so the returned count is a true
    POST count (the caller derives ``log_event_ids`` from it). Routing a
    multi-bit bitmap through here would post N x bits events while
    reporting N -- and BASIC-027 wants exactly N.

    Args:
        lfdi: Device LFDI (case-insensitive).
        alarm_status: Alarm bitmap with exactly one IEEE-mapped bit set.
        details: Optional human-readable detail string.
        interval: Seconds between each POST (0 = no delay).

    Returns:
        Number of events successfully posted.

    Raises:
        ValueError: If the device is unknown, has no log_event_list_href,
            or ``alarm_status`` does not resolve to exactly one mapped bit.
    """
    state = self.get_device_state(lfdi)
    if state is None:
        raise ValueError(f"Device {lfdi[:8]}... not found in telemetry manager")
    if not state.log_event_list_href:
        raise ValueError(f"Device {lfdi[:8]}... has no log_event_list_href")

    events = alarm_bits_to_log_events(0, alarm_status)
    if len(events) != 1:
        raise ValueError(
            f"LogEvent burst needs exactly one IEEE-mapped alarm bit; "
            f"alarm_status=0x{alarm_status:X} resolves to {len(events)} "
            f"(0 = no mapped bit set -- bits 11+ are reserved; "
            f">1 = multi-bit, which would post {len(events)} events per iteration)"
        )
    bit, code = events[0]

    posted = 0
    for i in range(self._LOG_EVENT_BURST_COUNT):
        if interval > 0 and i > 0:
            await asyncio.sleep(interval)
        try:
            await self._post_and_record_log_event(
                state, code, details=details or DER_ALARM_NAMES.get(bit)
            )
            posted += 1
        except Exception:
            logger.exception("LogEvent burst POST %d failed for %s", i, lfdi[:8])
            break

    logger.info(
        "LogEvent burst for %s: posted %d/%d events",
        lfdi[:8],
        posted,
        self._LOG_EVENT_BURST_COUNT,
    )
    return posted

find_device_with_log_events

find_device_with_log_events() -> str | None

Return the first active device LFDI that has a log_event_list_href.

Source code in src/py20305/telemetry/manager.py

def find_device_with_log_events(self) -> str | None:
    """Return the first active device LFDI that has a log_event_list_href."""
    for lfdi, state in self._devices.items():
        if state.log_event_list_href:
            return lfdi
    return None

get_device_state

get_device_state(lfdi: str) -> DeviceTelemetryState | None

Get telemetry state for a device.

Source code in src/py20305/telemetry/manager.py

def get_device_state(self, lfdi: str) -> DeviceTelemetryState | None:
    """Get telemetry state for a device."""
    return self._devices.get(lfdi.lower())

get_all_posted_log_events

get_all_posted_log_events() -> list[dict[str, Any]]

Return all posted log events across all devices.

Source code in src/py20305/telemetry/manager.py

def get_all_posted_log_events(self) -> list[dict[str, Any]]:
    """Return all posted log events across all devices."""
    events: list[dict[str, Any]] = []
    for state in self._devices.values():
        for ev in state.posted_log_events:
            events.append({**ev, "device": state.lfdi[:16]})
    events.sort(key=lambda e: e["timestamp"])
    return events

shutdown async

shutdown() -> None

Stop all metering tasks.

Source code in src/py20305/telemetry/manager.py

async def shutdown(self) -> None:
    """Stop all metering tasks."""
    await self._scheduler.cancel_all()
    self._devices.clear()
    logger.info("TelemetryManager shutdown complete")

py20305.telemetry.mup

MirrorUsagePoint creation for IEEE 2030.5 telemetry.

Creates MUP resources with system-wide MirrorMeterReadings (7 specs: W/Var/Hz/V/PF/VA/A) and, on multi-phase devices, an additional set of per-line readings for each phase advertised by the inverter (W/Var/V/PF/ VA/A per line; Hz is grid-wide and stays system-only).

ReadingTypeSpec dataclass

Specification for a MirrorMeterReading's ReadingType.

Source code in src/py20305/telemetry/mup.py

@dataclass(frozen=True, slots=True)
class ReadingTypeSpec:
    """Specification for a MirrorMeterReading's ReadingType."""

    key: str  # Metric key in monitoring data (W, Var, Hz, etc.)
    description: str  # Human-readable description
    commodity: int
    data_qualifier: int
    kind: int
    uom: int
    multiplier: int
    phase: int | None = None
    scale_func: Callable[[float | None], ScaledReading] | None = None
    # AccumulationBehaviourType: 12 = Instantaneous. Correct for every reading
    # we publish (all are point-in-time samples). Accumulated quantities -- e.g.
    # energy (Wh) -- would instead use 9 (Summation); override per-spec if added.
    accumulation_behaviour: int = 12

ReadingProfile

Bases: StrEnum

How a device's MirrorMeterReadings are encoded.

DEFAULT is the historical encoding. SIGNED_LOAD_CONVENTION carries the sign on every value (flowDirection 0) using the load convention -- active power negative on export, reactive power negative when the DER injects -- with commodity, dataQualifier and kind left Not Applicable, and adds line-to-line voltage and State of Energy readings when the device supplies them.

Source code in src/py20305/telemetry/mup.py

class ReadingProfile(StrEnum):
    """How a device's MirrorMeterReadings are encoded.

    ``DEFAULT`` is the historical encoding. ``SIGNED_LOAD_CONVENTION`` carries
    the sign on every value (flowDirection 0) using the load convention --
    active power negative on export, reactive power negative when the DER
    injects -- with commodity, dataQualifier and kind left Not Applicable, and
    adds line-to-line voltage and State of Energy readings when the device
    supplies them.
    """

    DEFAULT = "default"
    SIGNED_LOAD_CONVENTION = "signed_load_convention"

ac_type_of

ac_type_of(monitoring_data: dict[str, Any]) -> int | None

The SunSpec 701 ACType in monitoring data, or None if absent or not an int.

Source code in src/py20305/telemetry/mup.py

def ac_type_of(monitoring_data: dict[str, Any]) -> int | None:
    """The SunSpec 701 ``ACType`` in monitoring data, or None if absent or not an int."""
    ac_type = monitoring_data.get("ACType")
    if isinstance(ac_type, bool) or not isinstance(ac_type, int):
        return None
    return ac_type

registration_slots

registration_slots(monitoring_data: dict[str, Any], profile: ReadingProfile, registered: frozenset[int] = frozenset()) -> frozenset[int]

The mRID slots a MUP built now would register.

Under SIGNED_LOAD_CONVENTION the set grows when an optional reading first appears, which is the signal to re-POST the MUP so the new mRID arrives with its ReadingType (Rule h.2) instead of meeting Rule h.3.

Source code in src/py20305/telemetry/mup.py

def registration_slots(
    monitoring_data: dict[str, Any],
    profile: ReadingProfile,
    registered: frozenset[int] = frozenset(),
) -> frozenset[int]:
    """The mRID slots a MUP built now would register.

    Under SIGNED_LOAD_CONVENTION the set grows when an optional reading first
    appears, which is the signal to re-POST the MUP so the new mRID arrives
    with its ReadingType (Rule h.2) instead of meeting Rule h.3.
    """
    if profile is ReadingProfile.SIGNED_LOAD_CONVENTION:
        specs = _signed_load_specs_for(monitoring_data, registered)
    else:
        specs = _all_specs_for(monitoring_data)
    return frozenset(index for index, _ in specs)

compute_sfdi

compute_sfdi(lfdi: str) -> int

Compute SFDI from LFDI using Luhn-like check digit.

Parameters:

Name	Type	Description	Default
lfdi	str	40-character hex string (LFDI)	required

Returns:

Type	Description
int	SFDI as integer with check digit appended

Source code in src/py20305/telemetry/mup.py

def compute_sfdi(lfdi: str) -> int:
    """Compute SFDI from LFDI using Luhn-like check digit.

    Args:
        lfdi: 40-character hex string (LFDI)

    Returns:
        SFDI as integer with check digit appended
    """
    # Take first 9 hex characters and convert to decimal
    sfdi_base = lfdi[:9]
    sfdi_str = str(int(sfdi_base, 16))

    # Compute Luhn-like check digit
    digit_sum = sum(int(d) for d in sfdi_str)
    check_digit = (10 - (digit_sum % 10)) % 10

    return int(sfdi_str + str(check_digit))

create_mup

create_mup(lfdi: str, monitoring_data: dict[str, Any], post_rate: int, overrides: dict[str, ReadingOverride] | None = None, *, profile: ReadingProfile = ReadingProfile.DEFAULT, registered: frozenset[int] = frozenset()) -> MirrorUsagePoint

Create a MirrorUsagePoint with system + per-line MirrorMeterReadings.

Parameters:

Name	Type	Description	Default
lfdi	str	Device LFDI (will be normalized to lowercase internally, stored as uppercase in deviceLFDI per spec)	required
monitoring_data	dict[str, Any]	Raw monitoring values from connector.fetch_monitoring(). May include ACType (SunSpec 701 enum) and per-line keys (WL1, VL2 etc.); on a multi-phase device per-line readings are appended for every line advertised, including L1 (single-phase emits none -- see _all_specs_for).	required
post_rate	int	Posting rate in seconds	required
overrides	dict[str, ReadingOverride] | None	Optional per-key ReadingType overrides from the connector (see BaseConnector.reading_overrides); fields left unset keep the standard ReadingType metadata.	None
profile	ReadingProfile	How the readings are encoded (see :class:ReadingProfile).	DEFAULT
registered	frozenset[int]	Slots registered by an earlier POST of this MUP. Under SIGNED_LOAD_CONVENTION their optional readings are kept even when absent from monitoring_data.	frozenset()

Returns:

Type	Description
MirrorUsagePoint	MirrorUsagePoint with system + per-line meter readings.

Source code in src/py20305/telemetry/mup.py

def create_mup(
    lfdi: str,
    monitoring_data: dict[str, Any],
    post_rate: int,
    overrides: dict[str, ReadingOverride] | None = None,
    *,
    profile: ReadingProfile = ReadingProfile.DEFAULT,
    registered: frozenset[int] = frozenset(),
) -> MirrorUsagePoint:
    """Create a MirrorUsagePoint with system + per-line MirrorMeterReadings.

    Args:
        lfdi: Device LFDI (will be normalized to lowercase internally,
              stored as uppercase in deviceLFDI per spec)
        monitoring_data: Raw monitoring values from connector.fetch_monitoring().
              May include ``ACType`` (SunSpec 701 enum) and per-line
              keys (``WL1``, ``VL2`` etc.); on a multi-phase device per-line
              readings are appended for every line advertised, **including
              L1** (single-phase emits none -- see ``_all_specs_for``).
        post_rate: Posting rate in seconds
        overrides: Optional per-key ReadingType overrides from the connector
              (see ``BaseConnector.reading_overrides``); fields left unset keep
              the standard ReadingType metadata.
        profile: How the readings are encoded (see :class:`ReadingProfile`).
        registered: Slots registered by an earlier POST of this MUP. Under
              SIGNED_LOAD_CONVENTION their optional readings are kept even
              when absent from ``monitoring_data``.

    Returns:
        MirrorUsagePoint with system + per-line meter readings.
    """
    lfdi_norm = lfdi.lower()

    # Create MUP mRID (index 0)
    mup_mrid = _create_mrid(lfdi_norm, index=0)

    signed_load = profile is ReadingProfile.SIGNED_LOAD_CONVENTION
    specs = (
        _signed_load_specs_for(monitoring_data, registered)
        if signed_load
        else _all_specs_for(monitoring_data)
    )
    meter_readings: list[MirrorMeterReading1] = []
    for mrid_index, spec in specs:
        override = overrides.get(spec.key) if overrides else None
        flow_direction = (
            FLOW_NOT_APPLICABLE
            if signed_load
            else _scaled_for(spec, monitoring_data).flow_direction
        )
        reading_type = _create_reading_type(_effective_spec(spec, override), flow_direction)
        mmr_mrid = _create_mrid(lfdi_norm, index=mrid_index)
        mmr = MirrorMeterReading1(
            m_rid=mmr_mrid,
            description=spec.description,
            reading_type=reading_type,
        )
        meter_readings.append(mmr)

    # deviceLFDI is stored as uppercase hex bytes per spec
    device_lfdi_bytes = bytes.fromhex(lfdi_norm)

    return MirrorUsagePoint(
        m_rid=mup_mrid,
        description=f"DER {lfdi_norm[:8]}",
        version=VersionType(value=1),
        role_flags=RoleFlagsType(value=ROLE_FLAGS_DER.to_bytes(2, "big")),
        service_category_kind=ServiceKind(value=SERVICE_CATEGORY_ELECTRICITY),
        status=STATUS_ON,
        device_lfdi=device_lfdi_bytes,
        mirror_meter_reading=meter_readings,
        post_rate=post_rate,
    )

create_meter_reading_list

create_meter_reading_list(lfdi: str, monitoring_data: dict[str, Any], timestamp: int | None = None, overrides: dict[str, ReadingOverride] | None = None, *, post_rate: int = 300, next_update_time: int | None = None, stale: bool = False, profile: ReadingProfile = ReadingProfile.DEFAULT, registered: frozenset[int] | None = None) -> MirrorMeterReadingList

Create a MirrorMeterReadingList with current readings.

This is used when POSTing updated readings to an existing MUP. We post a reading only for quantities the connector actually supplied this cycle: a spec whose key is absent or None in monitoring_data is omitted rather than zero-filled, so an un-measured quantity doesn't show up as a misleading 0 (e.g. 0 V) on the server.

overrides (see BaseConnector.reading_overrides) supplies a per-key qualityFlags default; unset keys use the standard valid flag.

timestamp is the instant the values were acquired, which since the poll planner is no longer the instant they are posted. It becomes timePeriod.start and lastUpdateTime; per METER.001 the interval runs forward from start, so the existing duration semantics are unchanged and only the clock supplying start differs.

next_update_time is when this MirrorMeterReading will next be posted, not when the device will next be read, letting a server detect a stalled feed without waiting out the 72-hour MUP timeout. Post cadence rather than acquisition cadence because the two differ once the poll planner owns acquisition, and only the former is reliably knowable: a device in backoff has no predictable next read. None when the caller cannot say.

stale marks the sample older than its consumers asked for, or from an unreachable device. It ORs the questionable bit onto whatever quality the connector reported rather than replacing it, so a connector saying valid cannot mask staleness the client knows about.

This is a subset of what create_mup registered -- create_mup always registers the full ReadingType set for the device, so every omitted reading's ReadingType is still registered. Posting a subset of registered MirrorMeterReadings is valid per IEEE 2030.5 §10.11.3 (MirrorMeterReadings are optional in the MUP and updates), and mRIDs are deterministic from (lfdi, slot_index, pen) so a quantity that resumes reporting later matches its registered ReadingType (Rule h.1).

Under SIGNED_LOAD_CONVENTION every reading is an instantaneous sample, so its duration is 0 whatever the dataQualifier, and its value is scaled from the effective multiplier (see _signed_load_value). registered, when given, limits it to slots the MUP has registered, so a reading whose registration has not succeeded yet is held back rather than met with Rule h.3.

Source code in src/py20305/telemetry/mup.py

def create_meter_reading_list(
    lfdi: str,
    monitoring_data: dict[str, Any],
    timestamp: int | None = None,
    overrides: dict[str, ReadingOverride] | None = None,
    *,
    post_rate: int = 300,
    next_update_time: int | None = None,
    stale: bool = False,
    profile: ReadingProfile = ReadingProfile.DEFAULT,
    registered: frozenset[int] | None = None,
) -> MirrorMeterReadingList:
    """Create a MirrorMeterReadingList with current readings.

    This is used when POSTing updated readings to an existing MUP. We post a
    reading only for quantities the connector actually supplied this cycle: a
    spec whose key is absent or ``None`` in ``monitoring_data`` is omitted
    rather than zero-filled, so an un-measured quantity doesn't show up as a
    misleading ``0`` (e.g. ``0 V``) on the server.

    ``overrides`` (see ``BaseConnector.reading_overrides``) supplies a per-key
    ``qualityFlags`` default; unset keys use the standard ``valid`` flag.

    ``timestamp`` is the instant the values were *acquired*, which since the
    poll planner is no longer the instant they are posted. It becomes
    ``timePeriod.start`` and ``lastUpdateTime``; per METER.001 the interval runs
    forward from ``start``, so the existing duration semantics are unchanged and
    only the clock supplying ``start`` differs.

    ``next_update_time`` is when this MirrorMeterReading will next be *posted*,
    not when the device will next be read, letting a server detect a stalled
    feed without waiting out the 72-hour MUP timeout. Post cadence rather than
    acquisition cadence because the two differ once the poll planner owns
    acquisition, and only the former is reliably knowable: a device in backoff
    has no predictable next read. ``None`` when the caller cannot say.

    ``stale`` marks the sample older than its consumers asked for, or from an
    unreachable device. It ORs the *questionable* bit onto whatever quality the
    connector reported rather than replacing it, so a connector saying ``valid``
    cannot mask staleness the client knows about.

    This is a *subset* of what ``create_mup`` registered -- ``create_mup`` always
    registers the full ReadingType set for the device, so every omitted reading's
    ReadingType is still registered. Posting a subset of registered
    MirrorMeterReadings is valid per IEEE 2030.5 §10.11.3 (MirrorMeterReadings are
    optional in the MUP and updates), and mRIDs are deterministic from
    ``(lfdi, slot_index, pen)`` so a quantity that resumes reporting later matches
    its registered ReadingType (Rule h.1).

    Under ``SIGNED_LOAD_CONVENTION`` every reading is an instantaneous sample, so
    its duration is 0 whatever the dataQualifier, and its value is scaled from the
    effective multiplier (see ``_signed_load_value``). ``registered``, when given,
    limits it to slots the MUP has registered, so a reading whose registration
    has not succeeded yet is held back rather than met with Rule h.3.
    """
    if timestamp is None:
        timestamp = int(time.time())

    lfdi_norm = lfdi.lower()

    signed_load = profile is ReadingProfile.SIGNED_LOAD_CONVENTION
    specs = (
        _signed_load_specs_for(monitoring_data, frozenset())
        if signed_load
        else _all_specs_for(monitoring_data)
    )
    posted: dict[str, int] = {}
    meter_readings: list[MirrorMeterReading1] = []
    for mrid_index, spec in specs:
        if monitoring_data.get(spec.key) is None:
            continue
        if signed_load and registered is not None and mrid_index not in registered:
            continue
        mmr_mrid = _create_mrid(lfdi_norm, index=mrid_index)
        override = overrides.get(spec.key) if overrides else None
        # Optional per-cycle quality the connector attached to this sample.
        per_cycle_quality = monitoring_data.get(f"{spec.key}__quality")

        if signed_load:
            # Var precedes PF in slot order, system and per-line, so the sign
            # PF takes from it is already in `posted`.
            value = _signed_load_value(_effective_spec(spec, override), monitoring_data, posted)
            posted[spec.key] = value
            duration = 0
        else:
            value = _scaled_for(spec, monitoring_data).value
            # timePeriod duration: an interval reading (Average/Maximum/Minimum)
            # covers a window, and CSIP-AUS requires that window to match the MUP
            # postRate, so its duration is post_rate; an instantaneous reading is a
            # point in time -> duration 0. Use the effective dataQualifier (a
            # connector override can change it, e.g. print_demo reports W as
            # Maximum). getattr guards against a buggy in-process connector
            # returning a non-ReadingOverride value (the manager only checks the
            # top-level map).
            override_dq = (
                getattr(override, "data_qualifier", None) if override is not None else None
            )
            eff_dq = override_dq if override_dq is not None else spec.data_qualifier
            duration = 0 if eff_dq == _DATA_QUALIFIER_INSTANTANEOUS else post_rate
        reading = Reading1(
            quality_flags=_resolve_quality_flags(override, per_cycle_quality, stale=stale),
            time_period=DateTimeInterval(
                duration=duration,
                start=TimeType(value=timestamp),
            ),
            value=value,
        )

        # IEEE 10.11.3 rule n): ReadingType SHALL NOT be included in
        # subsequent POSTs -- only the initial MUP POST includes it.
        mmr = MirrorMeterReading1(
            m_rid=mmr_mrid,
            description=spec.description,
            reading=reading,
            # IEEE 2030.5 Annex B.17.1. Deliberately asymmetric, because the
            # two answer different questions: lastUpdateTime is when the data
            # behind this reading was refreshed (which the server cannot
            # otherwise derive), nextUpdateTime is when the mirror will next be
            # written (which is what makes a stalled feed detectable). Neither
            # is governed by a requirement in the conformance matrix; both are
            # optional and were previously unset, tracked as an IEEE Gap in
            # docs/spec/07-telemetry.md.
            last_update_time=TimeType(value=timestamp),
            next_update_time=(
                None if next_update_time is None else TimeType(value=next_update_time)
            ),
        )
        meter_readings.append(mmr)

    return MirrorMeterReadingList(
        mirror_meter_reading=meter_readings,
        all=len(meter_readings),
        results=len(meter_readings),
    )

py20305.telemetry.der_resource_manager

DerResourceManager for periodic PUT of DER resources.

Manages the lifecycle of DERCapability, DERSettings, and DERStatus PUT operations for each device. Follows the TelemetryManager composition pattern: owns a PollScheduler, accepts client + connector resolver.

DeviceDerState dataclass

Per-device DER resource tracking state.

Source code in src/py20305/telemetry/der_resource_manager.py

@dataclass
class DeviceDerState:
    """Per-device DER resource tracking state."""

    lfdi: str
    capability_href: str | None = None
    settings_href: str | None = None
    status_href: str | None = None
    scheduled_keys: list[str] = field(default_factory=list)
    # Change detection cache: only PUT when connector data changes
    last_settings_data: dict[str, Any] | None = None

DerResourceManager

Manages periodic PUT of DER resources for all devices.

Handles three resource types: - DERCapability (nameplate ratings from connector) - DERSettings (configuration from connector) - DERStatus (operational status from connector)

Each resource is PUT on its own poll cycle. Errors are logged but never propagate (the poll loop must not crash).

Source code in src/py20305/telemetry/der_resource_manager.py

class DerResourceManager:
    """Manages periodic PUT of DER resources for all devices.

    Handles three resource types:
    - DERCapability (nameplate ratings from connector)
    - DERSettings (configuration from connector)
    - DERStatus (operational status from connector)

    Each resource is PUT on its own poll cycle. Errors are logged
    but never propagate (the poll loop must not crash).
    """

    def __init__(
        self,
        client: Sep2Client,
        connector_resolver: Callable[[str], BaseConnector | Awaitable[BaseConnector]],
        command_observer: CommandObserver | None = None,
        device_telemetry: DeviceTelemetryEmitter | None = None,
    ) -> None:
        """Initialize the DerResourceManager.

        Args:
            client: IEEE 2030.5 HTTP client for PUT operations.
            connector_resolver: Callback to resolve LFDI -> connector.
                May be either sync (``Callable[[str], BaseConnector]``)
                or async (``Callable[[str], Awaitable[BaseConnector]]``);
                the cycle methods await the result iff it's awaitable.
                Async resolvers are preferred so first-touch construction
                of a real Modbus connector can offload its scan +
                readiness retries to a worker thread instead of blocking
                the event loop.
            command_observer: Offered the status this manager already fetches,
                as confirmation evidence for outstanding commands. Injected
                rather than imported for the same reason as on the dispatcher:
                the plane is a host-application concern and this is client code.
                Deliberately reuses the existing DERStatus read rather than
                adding one -- a second read of the same device to confirm a
                command would be a second acquisition path.
            device_telemetry: Reports the nameplate, configuration and status
                reads this manager issues to the monitoring system. These
                reads do not pass through a measurement source, so no other
                seam covers them. Optional and disabled by default.
        """
        self._client = client
        # Server timebase for telemetry timestamp defaults. isinstance guard
        # keeps AsyncMock clients in tests (whose .timebase is a Mock)
        # falling back to an identity timebase; production Sep2Client always
        # carries the real shared instance.
        tb = getattr(client, "timebase", None)
        self._timebase = tb if isinstance(tb, ServerTimebase) else ServerTimebase()
        self._resolve_connector = connector_resolver
        # See ConnectorDispatcher: `is None` so a falsy-when-empty observer is
        # not silently swapped for the no-op.
        self._commands = NullCommandObserver() if command_observer is None else command_observer
        self._telemetry = device_telemetry
        self._devices: dict[str, DeviceDerState] = {}
        self._scheduler = PollScheduler()

    async def _read(self, connector: Any, method: str, lfdi: str) -> dict[str, Any]:
        """Read one resource off a device, reporting it to the monitoring system.

        These reads do not go through a measurement source -- that path covers
        the metering poll only -- so without this they would be southbound
        traffic a monitoring system never sees, which is exactly the half the
        feature exists to close.

        Args:
            connector: The device connector to read from.
            method: The connector method to call, e.g. ``"fetch_status"``.
            lfdi: The device's LFDI, for attribution.

        Returns:
            Whatever the connector returned, unchanged.
        """
        values = await getattr(connector, method)()
        if self._telemetry is not None and isinstance(values, dict):
            self._telemetry.record_read(lfdi, values, connector=connector, lfdi=lfdi)
        return values  # type: ignore[no-any-return]

    def start_device(
        self,
        lfdi: str,
        capability_href: str | None,
        settings_href: str | None,
        status_href: str | None,
        *,
        capability_poll_rate: int = 86400,
        settings_poll_rate: int = 60,
        status_poll_rate: int = 300,
    ) -> None:
        """Start periodic DER resource PUTs for a device.

        Only schedules cycles for non-None hrefs. Each resource type runs
        at its own poll rate. DERCapability and DERSettings use change
        detection to skip PUTs when the data hasn't changed.

        Args:
            lfdi: Device LFDI.
            capability_href: Server path for DERCapability PUT.
            settings_href: Server path for DERSettings PUT.
            status_href: Server path for DERStatus PUT.
            capability_poll_rate: Interval for DERCapability PUTs (default 24h).
            settings_poll_rate: Interval for DERSettings PUTs (default 1min).
            status_poll_rate: Interval for DERStatus PUTs (default 300s).
        """
        lfdi_norm = lfdi.lower()
        state = DeviceDerState(
            lfdi=lfdi_norm,
            capability_href=capability_href,
            settings_href=settings_href,
            status_href=status_href,
        )
        self._devices[lfdi_norm] = state

        if capability_href:
            key = f"dercap_{lfdi_norm}"
            state.scheduled_keys.append(key)
            callback: Callable[[], Awaitable[None]] = partial(self._capability_cycle, lfdi_norm)
            self._scheduler.schedule(key, capability_poll_rate, callback)

        if settings_href:
            key = f"derset_{lfdi_norm}"
            state.scheduled_keys.append(key)
            callback = partial(self._settings_cycle, lfdi_norm)
            self._scheduler.schedule(key, settings_poll_rate, callback)

        if status_href:
            key = f"derstat_{lfdi_norm}"
            state.scheduled_keys.append(key)
            callback = partial(self._status_cycle, lfdi_norm)
            self._scheduler.schedule(key, status_poll_rate, callback)

        logger.info(
            "Started DER resources for %s (cap=%s@%ds, set=%s@%ds, stat=%s@%ds)",
            lfdi_norm[:8],
            bool(capability_href),
            capability_poll_rate,
            bool(settings_href),
            settings_poll_rate,
            bool(status_href),
            status_poll_rate,
        )

    async def _resolve_connector_async(self, lfdi: str) -> BaseConnector:
        """Resolve connector via the injected resolver.

        First-touch construction of a SunSpec connector runs a synchronous
        Modbus scan; an async resolver offloads that to a worker thread.
        Cache hits return without yielding the loop, so this stays cheap
        on every cycle after the first per-device call.

        Sync resolvers are still accepted for backward compat with
        embedded callers that wire in a custom callback -- the cycle
        awaits the result iff it's awaitable.
        """
        result = self._resolve_connector(lfdi)
        if inspect.isawaitable(result):
            return await result
        return result

    async def _capability_cycle(self, lfdi: str) -> None:
        """Single capability PUT cycle.

        Always PUTs regardless of whether the data changed since last cycle.
        DERCapability is a small, idempotent payload, and the server may have
        cleared it (e.g. between certification test runs).
        """
        state = self._devices.get(lfdi)
        if state is None or state.capability_href is None:
            return

        try:
            connector = await self._resolve_connector_async(lfdi)
            nameplate = await self._read(connector, "fetch_nameplate", lfdi)
            model = build_der_capability(
                nameplate,
                der_type=connector.der_type,
                csip_aus_mode=self._client.csip_aus_mode,
            )
            await self._client.put_bytes(
                state.capability_href,
                to_xml(
                    model,
                    server_2018_compat=self._client.server_2018_compat,
                    include_csipaus=self._client.csip_aus_mode,
                ),
            )
            logger.debug("DERCapability PUT for %s", lfdi[:8])
        except Sep2ProtocolError as e:
            from py20305.diagnostics import report

            hint = compat_hint_suffix(e, self._client.server_2018_compat)
            report(
                "warnings",
                f"DERCapability PUT for {lfdi[:8]} rejected (HTTP {e.status_code}): {e}{hint}",
                source="der_resource",
                dedup_key=f"der_put:{lfdi}:capability",
                details={
                    "lfdi": lfdi,
                    "resource": "DERCapability",
                    "status_code": e.status_code,
                    "error": str(e),
                },
            )
        except (Sep2ConnectionError, ConnectorError, OSError) as e:
            from py20305.diagnostics import report

            report(
                "warnings",
                f"DERCapability PUT failed for {lfdi[:8]}: {e}",
                source="der_resource",
                dedup_key=f"der_put:{lfdi}:capability",
                details={"lfdi": lfdi, "resource": "DERCapability", "error": str(e)},
            )
        except Exception as e:
            from py20305.diagnostics import report

            report(
                "warnings",
                f"Unexpected error in DERCapability PUT for {lfdi[:8]}: {e}",
                source="der_resource",
                dedup_key=f"der_put:{lfdi}:capability",
                details={"lfdi": lfdi, "resource": "DERCapability", "error": str(e)},
                exc_info=True,
            )

    async def _settings_cycle(self, lfdi: str) -> None:
        """Single settings PUT cycle (with change detection)."""
        state = self._devices.get(lfdi)
        if state is None or state.settings_href is None:
            return

        try:
            connector = await self._resolve_connector_async(lfdi)
            configuration = await self._read(connector, "fetch_configuration", lfdi)
            if configuration == state.last_settings_data:
                logger.debug("DERSettings unchanged for %s, skipping PUT", lfdi[:8])
                return
            try:
                model = build_der_settings(
                    configuration,
                    updated_time=int(self._timebase.now()),
                    csip_aus_mode=self._client.csip_aus_mode,
                )
            except ValueError as e:
                # The connector returned configuration missing a required field
                # (WMax / setMaxW is mandatory in the IEEE 2030.5 DERSettings
                # schema). This is a connector-contract problem, not an
                # unexpected crash -- report it as an actionable, deduplicated
                # warning instead of letting it fall through to the generic
                # handler that logs a full traceback every settings cycle.
                from py20305.diagnostics import report

                report(
                    "warnings",
                    f"DERSettings PUT skipped for {lfdi[:8]}: {e}. The connector's "
                    "fetch_configuration() must return a non-null WMax (nameplate "
                    "active power in watts); DERSettings cannot be built without it.",
                    source="der_resource",
                    dedup_key=f"der_put:{lfdi}:settings_config",
                    details={"lfdi": lfdi, "resource": "DERSettings", "error": str(e)},
                )
                return
            await self._client.put_bytes(
                state.settings_href,
                to_xml(
                    model,
                    server_2018_compat=self._client.server_2018_compat,
                    include_csipaus=self._client.csip_aus_mode,
                ),
            )
            state.last_settings_data = configuration
            logger.debug("DERSettings PUT for %s", lfdi[:8])
        except Sep2ProtocolError as e:
            from py20305.diagnostics import report

            hint = compat_hint_suffix(e, self._client.server_2018_compat)
            report(
                "warnings",
                f"DERSettings PUT for {lfdi[:8]} rejected (HTTP {e.status_code}): {e}{hint}",
                source="der_resource",
                dedup_key=f"der_put:{lfdi}:settings",
                details={
                    "lfdi": lfdi,
                    "resource": "DERSettings",
                    "status_code": e.status_code,
                    "error": str(e),
                },
            )
        except (Sep2ConnectionError, ConnectorError, OSError) as e:
            from py20305.diagnostics import report

            report(
                "warnings",
                f"DERSettings PUT failed for {lfdi[:8]}: {e}",
                source="der_resource",
                dedup_key=f"der_put:{lfdi}:settings",
                details={"lfdi": lfdi, "resource": "DERSettings", "error": str(e)},
            )
        except Exception as e:
            from py20305.diagnostics import report

            report(
                "warnings",
                f"Unexpected error in DERSettings PUT for {lfdi[:8]}: {e}",
                source="der_resource",
                dedup_key=f"der_put:{lfdi}:settings",
                details={"lfdi": lfdi, "resource": "DERSettings", "error": str(e)},
                exc_info=True,
            )

    async def _status_cycle(self, lfdi: str) -> None:
        """Single status PUT cycle."""
        state = self._devices.get(lfdi)
        if state is None or state.status_href is None:
            return

        try:
            connector = await self._resolve_connector_async(lfdi)
            # Stamped before the read: the plane discards an observation that
            # began before a command was issued, since it cannot reflect it.
            read_started_at = time.time()
            status_data: dict[str, Any] = await self._read(connector, "fetch_status", lfdi)
            self._commands.record_readback(lfdi, status_data, read_started_at=read_started_at)
            model = build_der_status(
                status_data,
                server_2018_compat=self._client.server_2018_compat,
                always_send_alarm_status=self._client.always_send_alarm_status,
                default_time=int(self._timebase.now()),
            )
            await self._client.put_bytes(
                state.status_href,
                to_xml(model, server_2018_compat=self._client.server_2018_compat),
            )
            logger.debug("DERStatus PUT for %s", lfdi[:8])
        except Sep2ProtocolError as e:
            from py20305.diagnostics import report

            hint = compat_hint_suffix(e, self._client.server_2018_compat)
            report(
                "warnings",
                f"DERStatus PUT for {lfdi[:8]} rejected (HTTP {e.status_code}): {e}{hint}",
                source="der_resource",
                dedup_key=f"der_put:{lfdi}:status",
                details={
                    "lfdi": lfdi,
                    "resource": "DERStatus",
                    "status_code": e.status_code,
                    "error": str(e),
                },
            )
        except (Sep2ConnectionError, ConnectorError, OSError) as e:
            from py20305.diagnostics import report

            report(
                "warnings",
                f"DERStatus PUT failed for {lfdi[:8]}: {e}",
                source="der_resource",
                dedup_key=f"der_put:{lfdi}:status",
                details={"lfdi": lfdi, "resource": "DERStatus", "error": str(e)},
            )
        except Exception as e:
            from py20305.diagnostics import report

            report(
                "warnings",
                f"Unexpected error in DERStatus PUT for {lfdi[:8]}: {e}",
                source="der_resource",
                dedup_key=f"der_put:{lfdi}:status",
                details={"lfdi": lfdi, "resource": "DERStatus", "error": str(e)},
                exc_info=True,
            )

    def stop_device(self, lfdi: str) -> None:
        """Stop the DER resource PUTs for one device.

        Drops the device's state, which is what the cycles read -- each becomes
        a no-op immediately, and the scheduler's tasks are cleaned up by
        ``shutdown``. Mirrors ``TelemetryManager.stop_metering``.

        Args:
            lfdi: Device LFDI.
        """
        self._devices.pop(lfdi.lower(), None)

    @property
    def active_devices(self) -> list[str]:
        """Return list of LFDIs with active DER resource cycles."""
        return list(self._devices.keys())

    async def shutdown(self) -> None:
        """Stop all scheduled DER resource tasks."""
        await self._scheduler.cancel_all()
        self._devices.clear()
        logger.info("DerResourceManager shutdown complete")

active_devices property

active_devices: list[str]

Return list of LFDIs with active DER resource cycles.

start_device

start_device(lfdi: str, capability_href: str | None, settings_href: str | None, status_href: str | None, *, capability_poll_rate: int = 86400, settings_poll_rate: int = 60, status_poll_rate: int = 300) -> None

Start periodic DER resource PUTs for a device.

Only schedules cycles for non-None hrefs. Each resource type runs at its own poll rate. DERCapability and DERSettings use change detection to skip PUTs when the data hasn't changed.

Parameters:

Name	Type	Description	Default
lfdi	str	Device LFDI.	required
capability_href	str | None	Server path for DERCapability PUT.	required
settings_href	str | None	Server path for DERSettings PUT.	required
status_href	str | None	Server path for DERStatus PUT.	required
capability_poll_rate	int	Interval for DERCapability PUTs (default 24h).	86400
settings_poll_rate	int	Interval for DERSettings PUTs (default 1min).	60
status_poll_rate	int	Interval for DERStatus PUTs (default 300s).	300

Source code in src/py20305/telemetry/der_resource_manager.py

def start_device(
    self,
    lfdi: str,
    capability_href: str | None,
    settings_href: str | None,
    status_href: str | None,
    *,
    capability_poll_rate: int = 86400,
    settings_poll_rate: int = 60,
    status_poll_rate: int = 300,
) -> None:
    """Start periodic DER resource PUTs for a device.

    Only schedules cycles for non-None hrefs. Each resource type runs
    at its own poll rate. DERCapability and DERSettings use change
    detection to skip PUTs when the data hasn't changed.

    Args:
        lfdi: Device LFDI.
        capability_href: Server path for DERCapability PUT.
        settings_href: Server path for DERSettings PUT.
        status_href: Server path for DERStatus PUT.
        capability_poll_rate: Interval for DERCapability PUTs (default 24h).
        settings_poll_rate: Interval for DERSettings PUTs (default 1min).
        status_poll_rate: Interval for DERStatus PUTs (default 300s).
    """
    lfdi_norm = lfdi.lower()
    state = DeviceDerState(
        lfdi=lfdi_norm,
        capability_href=capability_href,
        settings_href=settings_href,
        status_href=status_href,
    )
    self._devices[lfdi_norm] = state

    if capability_href:
        key = f"dercap_{lfdi_norm}"
        state.scheduled_keys.append(key)
        callback: Callable[[], Awaitable[None]] = partial(self._capability_cycle, lfdi_norm)
        self._scheduler.schedule(key, capability_poll_rate, callback)

    if settings_href:
        key = f"derset_{lfdi_norm}"
        state.scheduled_keys.append(key)
        callback = partial(self._settings_cycle, lfdi_norm)
        self._scheduler.schedule(key, settings_poll_rate, callback)

    if status_href:
        key = f"derstat_{lfdi_norm}"
        state.scheduled_keys.append(key)
        callback = partial(self._status_cycle, lfdi_norm)
        self._scheduler.schedule(key, status_poll_rate, callback)

    logger.info(
        "Started DER resources for %s (cap=%s@%ds, set=%s@%ds, stat=%s@%ds)",
        lfdi_norm[:8],
        bool(capability_href),
        capability_poll_rate,
        bool(settings_href),
        settings_poll_rate,
        bool(status_href),
        status_poll_rate,
    )

stop_device

stop_device(lfdi: str) -> None

Stop the DER resource PUTs for one device.

Drops the device's state, which is what the cycles read -- each becomes a no-op immediately, and the scheduler's tasks are cleaned up by shutdown. Mirrors TelemetryManager.stop_metering.

Parameters:

Name	Type	Description	Default
lfdi	str	Device LFDI.	required

Source code in src/py20305/telemetry/der_resource_manager.py

def stop_device(self, lfdi: str) -> None:
    """Stop the DER resource PUTs for one device.

    Drops the device's state, which is what the cycles read -- each becomes
    a no-op immediately, and the scheduler's tasks are cleaned up by
    ``shutdown``. Mirrors ``TelemetryManager.stop_metering``.

    Args:
        lfdi: Device LFDI.
    """
    self._devices.pop(lfdi.lower(), None)

shutdown async

shutdown() -> None

Stop all scheduled DER resource tasks.

Source code in src/py20305/telemetry/der_resource_manager.py

async def shutdown(self) -> None:
    """Stop all scheduled DER resource tasks."""
    await self._scheduler.cancel_all()
    self._devices.clear()
    logger.info("DerResourceManager shutdown complete")

py20305.readings

Measurement vocabulary, and the source interface that supplies it.

One acquisition of a device's monitoring data becomes one :class:DeviceSnapshot holding a :class:PointEntry per monitoring key. Entries are immutable: an acquisition replaces a device's snapshot rather than mutating it, so a reader that holds a snapshot always sees a coherent set of points from a single device read.

:class:MeasurementSource is where a consumer gets those snapshots. It exists so the telemetry path states what it needs -- a reading for this device, no older than this -- without naming who satisfies it. :class:DirectConnectorSource is the answer for a deployment with one consumer: read the connector, cache nothing. A deployment serving several upstream interfaces over one device model supplies a caching, demand-coalescing implementation instead, and nothing in the telemetry path changes.

Quality

Bases: Enum

The store's own health model for a value.

Deliberately not a protocol quality value. IEEE 2030.5 qualityFlags is a separate bitfield the connector supplies and the MirrorUsagePoint path posts; it rides along on :attr:PointEntry.protocol_quality rather than being mapped onto this enum. The two answer different questions -- "did the read succeed" versus "is this value still current".

Source code in src/py20305/readings.py

class Quality(Enum):
    """The store's own health model for a value.

    Deliberately *not* a protocol quality value. IEEE 2030.5 ``qualityFlags``
    is a separate bitfield the connector supplies and the MirrorUsagePoint path
    posts; it rides along on :attr:`PointEntry.protocol_quality` rather than
    being mapped onto this enum. The two answer different questions -- "did the
    read succeed" versus "is this value still current".
    """

    GOOD = "good"
    #: Acquired successfully but older than the tightest demand asked for.
    #: Applied by ``with_freshness`` at read time rather than stored, because
    #: the same sample is fresh to one consumer and stale to another.
    STALE = "stale"
    COMM_LOST = "comm_lost"

PointEntry dataclass

One monitoring key's value and everything acquired alongside it.

Source code in src/py20305/readings.py

@dataclass(frozen=True, slots=True)
class PointEntry:
    """One monitoring key's value and everything acquired alongside it."""

    value: Any
    #: Wall-clock epoch seconds at which the acquisition that produced this
    #: value *began*. Not read-completion time: on a multi-second Modbus scan
    #: the two differ, and stamping completion would make the sample look
    #: fresher than it is. The true per-register instant is unknowable, so this
    #: is the closest available answer to "when the device produced it" that
    #: errs old rather than new.
    #: For a COMM_LOST entry it stays at the last *successful* acquisition, so
    #: the age a reader sees grows while the device is unreachable.
    source_timestamp: float
    quality: Quality
    #: The connector's per-cycle ``"<key>__quality"``, carried verbatim. None
    #: when the connector supplied none for this key this cycle.
    protocol_quality: int | None

DeviceSnapshot dataclass

A device's points as of one acquisition.

age is deliberately absent: it depends on a clock, and the store has none. Callers compute it from :attr:last_success against whatever clock they already hold.

Source code in src/py20305/readings.py

@dataclass(frozen=True, slots=True)
class DeviceSnapshot:
    """A device's points as of one acquisition.

    ``age`` is deliberately absent: it depends on a clock, and the store has
    none. Callers compute it from :attr:`last_success` against whatever clock
    they already hold.
    """

    entries: Mapping[str, PointEntry]
    #: The connector's ``reading_overrides()`` as of this acquisition, keyed by
    #: monitoring key. Device-level rather than per-entry because it legitimately
    #: covers keys absent from this cycle's values: ``create_mup`` registers the
    #: *full* ReadingType set, so an override for a quantity the connector isn't
    #: reporting right now still shapes what gets registered. Read after the
    #: values because an out-of-process connector populates it from the same
    #: ``fetch_monitoring`` response.
    reading_overrides: Mapping[str, ReadingOverride]
    #: Store-wide monotonic counter identifying the acquisition these entries
    #: came from. Orders acquisitions across devices.
    sequence: int
    #: Epoch seconds of the last acquisition that succeeded, or None if none
    #: ever has. Distinct from a COMM_LOST entry's ``source_timestamp`` only
    #: when the device has never been read.
    last_success: float | None
    quality: Quality
    #: What the failed acquisition raised, when quality is COMM_LOST. Carried
    #: rather than reported here so each consumer keeps its own operator-facing
    #: identity -- the metering cycle and the management API describe the same
    #: failure differently, and the store has no business choosing between them.
    error: Exception | None = None

MeasurementSource

Bases: Protocol

Where the telemetry path gets a device's measurements.

Three operations, because a consumer's interest in a device has a lifetime: it announces what freshness it needs, reads repeatedly, and eventually stops caring. An implementation that coalesces demand across several consumers needs the first and last to do its job; one serving a single consumer can ignore them, which is exactly what :class:DirectConnectorSource does.

Source code in src/py20305/readings.py

@runtime_checkable
class MeasurementSource(Protocol):
    """Where the telemetry path gets a device's measurements.

    Three operations, because a consumer's interest in a device has a lifetime:
    it announces what freshness it needs, reads repeatedly, and eventually stops
    caring. An implementation that coalesces demand across several consumers
    needs the first and last to do its job; one serving a single consumer can
    ignore them, which is exactly what :class:`DirectConnectorSource` does.
    """

    def declare(self, device: str, max_age: float) -> None:
        """Announce that this consumer needs ``device`` no staler than ``max_age``."""
        ...

    def release(self, device: str) -> None:
        """Withdraw interest in ``device``; any cached state for it may be dropped."""
        ...

    async def read(self, device: str, *, max_age: float | None = None) -> DeviceSnapshot:
        """Return the best available snapshot of ``device``, judged against ``max_age``.

        ``max_age`` is the freshness the caller wants, not a guarantee it gets:
        a source may return an older sample marked ``STALE``, or a ``COMM_LOST``
        one carrying the last known values. What it will not do is return a
        stale sample looking fresh -- the verdict is already applied, so callers
        do not call :func:`with_freshness` themselves. ``None`` means the caller
        accepts whatever the source considers current.
        """
        ...

declare

declare(device: str, max_age: float) -> None

Announce that this consumer needs device no staler than max_age.

Source code in src/py20305/readings.py

def declare(self, device: str, max_age: float) -> None:
    """Announce that this consumer needs ``device`` no staler than ``max_age``."""
    ...

release

release(device: str) -> None

Withdraw interest in device; any cached state for it may be dropped.

Source code in src/py20305/readings.py

def release(self, device: str) -> None:
    """Withdraw interest in ``device``; any cached state for it may be dropped."""
    ...

read async

read(device: str, *, max_age: float | None = None) -> DeviceSnapshot

Return the best available snapshot of device, judged against max_age.

max_age is the freshness the caller wants, not a guarantee it gets: a source may return an older sample marked STALE, or a COMM_LOST one carrying the last known values. What it will not do is return a stale sample looking fresh -- the verdict is already applied, so callers do not call :func:with_freshness themselves. None means the caller accepts whatever the source considers current.

Source code in src/py20305/readings.py

async def read(self, device: str, *, max_age: float | None = None) -> DeviceSnapshot:
    """Return the best available snapshot of ``device``, judged against ``max_age``.

    ``max_age`` is the freshness the caller wants, not a guarantee it gets:
    a source may return an older sample marked ``STALE``, or a ``COMM_LOST``
    one carrying the last known values. What it will not do is return a
    stale sample looking fresh -- the verdict is already applied, so callers
    do not call :func:`with_freshness` themselves. ``None`` means the caller
    accepts whatever the source considers current.
    """
    ...

DirectConnectorSource

Reads the connector on every request; never serves one from cache.

The single-consumer answer. With one upstream interface there is no second reader to share a cached value with and no competing cadence to coalesce, so caching a read would add staleness without removing a device read. declare and release are deliberately inert: freshness is whatever the read just returned.

It does retain the last successful snapshot per device. That is history, not a cache -- it is never returned in place of a read, only used so a failed read can report comm-lost against the last known values instead of an empty set, matching what the store-backed path does.

Source code in src/py20305/readings.py

class DirectConnectorSource:
    """Reads the connector on every request; never serves one from cache.

    The single-consumer answer. With one upstream interface there is no second
    reader to share a cached value with and no competing cadence to coalesce, so
    caching a *read* would add staleness without removing a device read.
    ``declare`` and ``release`` are deliberately inert: freshness is whatever the
    read just returned.

    It does retain the last successful snapshot per device. That is history, not
    a cache -- it is never returned in place of a read, only used so a failed
    read can report comm-lost against the last known values instead of an empty
    set, matching what the store-backed path does.
    """

    def __init__(
        self,
        connector_resolver: Callable[[str], BaseConnector | Awaitable[BaseConnector]],
        *,
        clock: Callable[[], float] = time.time,
        telemetry: DeviceTelemetryEmitter | None = None,
    ) -> None:
        """
        Args:
            connector_resolver: LFDI -> connector. **Must be memoizing**: this
                source resolves on every read, and a caller may resolve the same
                device separately in the same cycle for other work. A resolver
                that constructs on each call would repeat first-touch setup -- a
                Modbus scan plus readiness retries -- and could hand out two
                instances for one device in one cycle. Both real resolvers
                satisfy this: the client's goes through
                ``LazyConnectorProxy.aresolve``, which has a cache fast-path,
                and an embedded caller typically closes over a single instance.
                ``AcquisitionService`` carries the same requirement, so this is
                the existing contract stated rather than a new one.
            clock: Source of acquisition timestamps. Injectable so tests can
                assert on age without sleeping.
            telemetry: Reports each read to the monitoring system. Optional and
                disabled by default; omitted, reads behave exactly as before.
        """
        self._resolve_connector = connector_resolver
        self._clock = clock
        self._telemetry = telemetry
        self._sequence = 0
        #: Last successful read per device, so a failed read can report
        #: comm-lost against real history rather than as a never-read device.
        self._last: dict[str, DeviceSnapshot] = {}

    def declare(self, device: str, max_age: float) -> None:
        """No-op: a single consumer's demand is the ``max_age`` it passes to read."""

    def release(self, device: str) -> None:
        """Drop the retained last-good snapshot for ``device``."""
        self._last.pop(device, None)

    async def read(self, device: str, *, max_age: float | None = None) -> DeviceSnapshot:
        """Read ``device`` through its connector, now."""
        self._sequence += 1
        try:
            resolved = self._resolve_connector(device)
            connector = await resolved if inspect.isawaitable(resolved) else resolved
            # Stamped after resolution and before the read, matching
            # AcquisitionService. First-touch construction of a real connector
            # can take seconds (a Modbus scan plus readiness retries); counting
            # that against the sample would report it older than it is. Stamping
            # completion instead would report it fresher, which is the more
            # dangerous direction for a field that drives staleness.
            started = self._clock()
            values = await connector.fetch_monitoring()
            # A connector is third-party code and may return anything. Coerce a
            # non-dict override map away, and drop individual values that aren't
            # ReadingOverride, rather than letting either reach the
            # MirrorUsagePoint path; reject a non-dict value map outright --
            # there is no reading to salvage from it.
            raw_overrides = connector.reading_overrides()
            overrides = typed_overrides(raw_overrides if isinstance(raw_overrides, dict) else None)
            if not isinstance(values, dict):
                raise TypeError(f"fetch_monitoring returned {type(values).__name__}, expected dict")
        except Exception as exc:
            # Broad by intent. A connector is third-party code, including
            # out-of-process connectors, so any failure has to degrade to
            # comm-lost rather than kill the caller's cycle. ConnectorError and
            # OSError are the expected shapes; anything else still means "no
            # reading this time" and is carried on the snapshot for the
            # consumer to report in its own words.
            previous = self._last.get(device)
            # Each entry is rebuilt COMM_LOST rather than carried over as-is.
            # Keeping the value and its original source_timestamp is deliberate
            # -- the age a reader sees grows through the outage -- but leaving
            # the per-entry quality at GOOD under a COMM_LOST snapshot would
            # contradict both the snapshot and PointStore.record_failure, which
            # any reader inspecting entries individually would then disagree
            # with.
            carried = {
                key: PointEntry(
                    value=entry.value,
                    source_timestamp=entry.source_timestamp,
                    quality=Quality.COMM_LOST,
                    protocol_quality=entry.protocol_quality,
                )
                for key, entry in (previous.entries if previous else {}).items()
            }
            return DeviceSnapshot(
                entries=MappingProxyType(carried),
                reading_overrides=previous.reading_overrides if previous else MappingProxyType({}),
                sequence=self._sequence,
                last_success=previous.last_success if previous else None,
                quality=Quality.COMM_LOST,
                error=exc,
            )

        measured, protocol_quality = split_quality(values)
        entries = {
            key: PointEntry(
                value=value,
                source_timestamp=started,
                quality=Quality.GOOD,
                protocol_quality=protocol_quality.get(key),
            )
            for key, value in measured.items()
        }
        snapshot = DeviceSnapshot(
            entries=MappingProxyType(entries),
            reading_overrides=MappingProxyType(dict(overrides or {})),
            sequence=self._sequence,
            last_success=started,
            quality=Quality.GOOD,
            error=None,
        )
        self._last[device] = snapshot
        # After the snapshot, not before: telemetry reports a read that
        # actually produced one, and a read that raised produced no reading.
        if self._telemetry is not None:
            self._telemetry.record_read(device, values, connector=connector)
        return snapshot

declare

declare(device: str, max_age: float) -> None

No-op: a single consumer's demand is the max_age it passes to read.

Source code in src/py20305/readings.py

def declare(self, device: str, max_age: float) -> None:
    """No-op: a single consumer's demand is the ``max_age`` it passes to read."""

release

release(device: str) -> None

Drop the retained last-good snapshot for device.

Source code in src/py20305/readings.py

def release(self, device: str) -> None:
    """Drop the retained last-good snapshot for ``device``."""
    self._last.pop(device, None)

read async

read(device: str, *, max_age: float | None = None) -> DeviceSnapshot

Read device through its connector, now.

Source code in src/py20305/readings.py

async def read(self, device: str, *, max_age: float | None = None) -> DeviceSnapshot:
    """Read ``device`` through its connector, now."""
    self._sequence += 1
    try:
        resolved = self._resolve_connector(device)
        connector = await resolved if inspect.isawaitable(resolved) else resolved
        # Stamped after resolution and before the read, matching
        # AcquisitionService. First-touch construction of a real connector
        # can take seconds (a Modbus scan plus readiness retries); counting
        # that against the sample would report it older than it is. Stamping
        # completion instead would report it fresher, which is the more
        # dangerous direction for a field that drives staleness.
        started = self._clock()
        values = await connector.fetch_monitoring()
        # A connector is third-party code and may return anything. Coerce a
        # non-dict override map away, and drop individual values that aren't
        # ReadingOverride, rather than letting either reach the
        # MirrorUsagePoint path; reject a non-dict value map outright --
        # there is no reading to salvage from it.
        raw_overrides = connector.reading_overrides()
        overrides = typed_overrides(raw_overrides if isinstance(raw_overrides, dict) else None)
        if not isinstance(values, dict):
            raise TypeError(f"fetch_monitoring returned {type(values).__name__}, expected dict")
    except Exception as exc:
        # Broad by intent. A connector is third-party code, including
        # out-of-process connectors, so any failure has to degrade to
        # comm-lost rather than kill the caller's cycle. ConnectorError and
        # OSError are the expected shapes; anything else still means "no
        # reading this time" and is carried on the snapshot for the
        # consumer to report in its own words.
        previous = self._last.get(device)
        # Each entry is rebuilt COMM_LOST rather than carried over as-is.
        # Keeping the value and its original source_timestamp is deliberate
        # -- the age a reader sees grows through the outage -- but leaving
        # the per-entry quality at GOOD under a COMM_LOST snapshot would
        # contradict both the snapshot and PointStore.record_failure, which
        # any reader inspecting entries individually would then disagree
        # with.
        carried = {
            key: PointEntry(
                value=entry.value,
                source_timestamp=entry.source_timestamp,
                quality=Quality.COMM_LOST,
                protocol_quality=entry.protocol_quality,
            )
            for key, entry in (previous.entries if previous else {}).items()
        }
        return DeviceSnapshot(
            entries=MappingProxyType(carried),
            reading_overrides=previous.reading_overrides if previous else MappingProxyType({}),
            sequence=self._sequence,
            last_success=previous.last_success if previous else None,
            quality=Quality.COMM_LOST,
            error=exc,
        )

    measured, protocol_quality = split_quality(values)
    entries = {
        key: PointEntry(
            value=value,
            source_timestamp=started,
            quality=Quality.GOOD,
            protocol_quality=protocol_quality.get(key),
        )
        for key, value in measured.items()
    }
    snapshot = DeviceSnapshot(
        entries=MappingProxyType(entries),
        reading_overrides=MappingProxyType(dict(overrides or {})),
        sequence=self._sequence,
        last_success=started,
        quality=Quality.GOOD,
        error=None,
    )
    self._last[device] = snapshot
    # After the snapshot, not before: telemetry reports a read that
    # actually produced one, and a read that raised produced no reading.
    if self._telemetry is not None:
        self._telemetry.record_read(device, values, connector=connector)
    return snapshot

split_quality

split_quality(values: dict[str, Any]) -> tuple[dict[str, Any], dict[str, int]]

Separate measured keys from their "<key>__quality" companions.

Lives with the vocabulary rather than with any store because it encodes the connector convention itself, and every path that turns a raw fetch_monitoring response into entries has to apply it identically. A second copy would drift.

Source code in src/py20305/readings.py

def split_quality(values: dict[str, Any]) -> tuple[dict[str, Any], dict[str, int]]:
    """Separate measured keys from their ``"<key>__quality"`` companions.

    Lives with the vocabulary rather than with any store because it encodes the
    connector convention itself, and every path that turns a raw
    ``fetch_monitoring`` response into entries has to apply it identically. A
    second copy would drift.
    """
    measured: dict[str, Any] = {}
    quality: dict[str, int] = {}
    for key, value in values.items():
        if key.endswith(QUALITY_SUFFIX):
            base = key[: -len(QUALITY_SUFFIX)]
            # A non-int here would be a connector bug. Drop it rather than
            # carrying it: the MUP path validates qualityFlags again and would
            # log the same complaint once per cycle forever. bool is excluded
            # explicitly -- it is an int subclass, and True would post as 1.
            if isinstance(value, int) and not isinstance(value, bool):
                quality[base] = value
        else:
            measured[key] = value
    return measured, quality

typed_overrides

typed_overrides(overrides: dict[str, Any] | None) -> dict[str, ReadingOverride] | None

Keep only well-formed :class:ReadingOverride values from a connector's dict.

Here for the same reason as :func:split_quality: it encodes the connector contract, and every path that turns a raw reading_overrides() response into a snapshot has to apply it identically or the snapshots disagree. :attr:DeviceSnapshot.reading_overrides is typed Mapping[str, ReadingOverride], and the values arrive from third-party code through an Any, so nothing but this check keeps the annotation honest.

Source code in src/py20305/readings.py

def typed_overrides(overrides: dict[str, Any] | None) -> dict[str, ReadingOverride] | None:
    """Keep only well-formed :class:`ReadingOverride` values from a connector's dict.

    Here for the same reason as :func:`split_quality`: it encodes the connector
    contract, and every path that turns a raw ``reading_overrides()`` response
    into a snapshot has to apply it identically or the snapshots disagree.
    :attr:`DeviceSnapshot.reading_overrides` is typed
    ``Mapping[str, ReadingOverride]``, and the values arrive from third-party
    code through an ``Any``, so nothing but this check keeps the annotation
    honest.
    """
    if overrides is None:
        return None
    return {k: v for k, v in overrides.items() if isinstance(v, ReadingOverride)}

with_freshness

with_freshness(snapshot: DeviceSnapshot, max_age: float | None, now: float) -> DeviceSnapshot

Downgrade a snapshot's GOOD entries to STALE once they exceed max_age.

Kept out of the store because staleness is not a property of the data: the same sample is fresh to a consumer wanting minute-old values and stale to one wanting second-old values. The store records when a value was acquired; what counts as too old belongs to whoever declared the demand.

COMM_LOST is left alone -- it already says something stronger than stale, and a value from an unreachable device is not merely late.

Source code in src/py20305/readings.py

def with_freshness(snapshot: DeviceSnapshot, max_age: float | None, now: float) -> DeviceSnapshot:
    """Downgrade a snapshot's GOOD entries to STALE once they exceed ``max_age``.

    Kept out of the store because staleness is not a property of the data: the
    same sample is fresh to a consumer wanting minute-old values and stale to
    one wanting second-old values. The store records when a value was acquired;
    what counts as too old belongs to whoever declared the demand.

    COMM_LOST is left alone -- it already says something stronger than stale,
    and a value from an unreachable device is not merely late.
    """
    if max_age is None or snapshot.quality is not Quality.GOOD:
        return snapshot
    if snapshot.last_success is None or now - snapshot.last_success <= max_age:
        return snapshot

    entries = {
        key: (
            entry
            if entry.quality is not Quality.GOOD
            else PointEntry(
                value=entry.value,
                source_timestamp=entry.source_timestamp,
                quality=Quality.STALE,
                protocol_quality=entry.protocol_quality,
            )
        )
        for key, entry in snapshot.entries.items()
    }
    return DeviceSnapshot(
        entries=MappingProxyType(entries),
        reading_overrides=snapshot.reading_overrides,
        sequence=snapshot.sequence,
        last_success=snapshot.last_success,
        quality=Quality.STALE,
        error=snapshot.error,
    )