servomexlib¶

async def analyser_status(self, *, timeout: float | None = None) -> AnalyserStatus:
    """Return the analyser-level status."""
    return await self._session.analyser_status(timeout=timeout)

async def calibration_status(
    self, group: int = 1, *, timeout: float | None = None
) -> CalibrationProgress:
    """Return autocalibration progress for ``group`` (Modbus only)."""
    return await self._session.calibration_status(group, timeout=timeout)

async def identify(self, *, timeout: float | None = None) -> DeviceInfo:
    """Return device identity, caching it on first call."""
    self._info = await self._session.identify(timeout=timeout)
    return self._info

async def poll(self, *, wait_fresh: bool = False, timeout: float | None = None) -> Frame:
    """Return one frame — all channels + analyser status, one tick.

    ``wait_fresh=True`` waits for the *next* continuous broadcast instead of
    returning the cached latest; it is a no-op for the polled Modbus path.
    """
    return await self._session.read_frame(wait_fresh=wait_fresh, timeout=timeout)

async def poll_samples(
    self, *, names: Sequence[str] | None = None, timeout: float | None = None
) -> list[Sample]:
    """Poll one frame and fan it out into long-format samples (``PollSource``).

    Makes the analyser a :class:`~servomexlib.streaming.poll_source.PollSource`
    the recorder can drive. ``names`` is ignored for a solo analyser (it is the
    manager's device-subset selector).
    """
    del names
    frame = await self._session.read_frame(timeout=timeout)
    return frame.as_samples(device=self._device)

async def read_all(self, *, timeout: float | None = None) -> dict[ChannelId, Reading]:
    """Return every channel's latest reading keyed by id."""
    frame = await self._session.read_frame(timeout=timeout)
    return {reading.channel: reading for reading in frame.readings}

async def read_channel(
    self, channel: ChannelId | str, *, timeout: float | None = None
) -> Reading:
    """Return one channel's latest reading."""
    return await self._session.read_channel(_coerce_channel(channel), timeout=timeout)

def snapshot(self) -> Frame:
    """Return the cached latest frame without any I/O."""
    return self._session.snapshot()

async def start_calibration(
    self, group: int, *, confirm: bool = False, timeout: float | None = None
) -> None:
    """Start autocalibration for ``group`` (Modbus only; requires ``confirm=True``)."""
    await self._session.start_calibration(group, confirm=confirm, timeout=timeout)

async def status(
    self, channel: ChannelId | str, *, timeout: float | None = None
) -> ChannelStatus:
    """Return one channel's latest status."""
    return await self._session.status(_coerce_channel(channel), timeout=timeout)

async def stop_calibration(
    self, *, confirm: bool = False, timeout: float | None = None
) -> None:
    """Stop all autocalibration (Modbus only; requires ``confirm=True``)."""
    await self._session.stop_calibration(confirm=confirm, timeout=timeout)

def stream(
    self, *, mode: StreamMode | None = None, rate_hz: float | None = None
) -> StreamingSession:
    """Stream samples — ``AUTOPRINT`` (continuous) or ``POLL`` (Modbus).

    Defaults the mode per protocol. Continuous mode subscribes passively to the
    unsolicited broadcast (``rate_hz`` is ignored — the analyser sets the
    cadence). Modbus mode drives the drift-free
    :func:`~servomexlib.streaming.recorder.record` loop at ``rate_hz``
    (default ``1.0``) inside a task group the returned session owns.

    Use it as an async context manager so the poll loop starts and stops with
    the session::

        async with anz.stream(rate_hz=2) as samples:
            async for sample in samples:
                ...
    """
    client_stream = getattr(self._client, "stream", None)
    if callable(client_stream):  # continuous: passive broadcast subscribe
        return cast("StreamingSession", client_stream(mode=mode))
    resolved = mode if mode is not None else StreamMode.POLL
    if resolved is not StreamMode.POLL:
        raise ServomexValidationError(
            f"{self.protocol.value} mode only supports POLL streaming, not {resolved.value}",
        )
    return self._poll_stream(rate_hz if rate_hz is not None else 1.0)

@classmethod
def failure(cls, error: ServomexError) -> Self:
    """Build a failure result wrapping ``error``."""
    return cls(value=None, error=error)

@classmethod
def success(cls, value: T) -> Self:
    """Build a success result wrapping ``value``."""
    return cls(value=value, error=None)

def merged(self, **updates: Any) -> Self:
    """Return a new context with ``updates`` overlaid. Unknown keys go to ``extra``."""
    known: dict[str, Any] = {}
    extra_updates: dict[str, Any] = {}
    for key, value in updates.items():
        if key in _CONTEXT_KNOWN_FIELDS:
            known[key] = value
        else:
            extra_updates[key] = value

    new_extra: Mapping[str, Any] = (
        MappingProxyType({**self.extra, **extra_updates}) if extra_updates else self.extra
    )
    return replace(self, **known, extra=new_extra)

def add_script(self, request: bytes, reply: ScriptedReply) -> None:
    """Register or overwrite the canned reply for ``request``."""
    self._script[bytes(request)] = reply

async def emit(self, frames: Sequence[bytes], *, interval: float = 0.0) -> None:
    """Feed ``frames`` one at a time, optionally ``interval`` seconds apart.

    Run as a background task to simulate a continuous broadcaster:
    ``task_group.start_soon(lambda: fake.emit(frames, interval=0.05))``.
    """
    for frame in frames:
        if interval:
            await anyio.sleep(interval)
        self.feed(frame)

def feed(self, data: bytes) -> None:
    """Push unsolicited bytes into the inbound buffer."""
    self._inbound += data
    self._wake()

def as_samples(self, *, device: str = "") -> list[Sample]:
    """Fan the frame out into one long-format :class:`Sample` per reading."""
    return [
        Sample(
            device=device,
            channel=reading.channel,
            reading=reading,
            protocol=self.protocol,
            monotonic_ns=self.monotonic_ns,
            received_at=self.received_at,
        )
        for reading in self.readings
    ]

def channel(self, cid: ChannelId) -> Reading:
    """Return the :class:`Reading` for ``cid``.

    Raises:
        ServomexValidationError: ``cid`` is not present in this frame.
    """
    for reading in self.readings:
        if reading.channel == cid:
            return reading
    raise ServomexValidationError(
        f"channel {getattr(cid, 'value', cid)} not present in frame",
    )

async def poll_samples(
    self, *, names: Sequence[str] | None = None, timeout: float | None = None
) -> Sequence[Sample]:
    """Read this tick's samples across every channel (and managed device).

    Args:
        names: Subset of device names to poll (manager only); ``None`` polls
            everything. A solo analyser ignores this.
        timeout: Per-poll I/O ceiling, or ``None`` for the source default.

    Returns:
        A flat sequence of :class:`Sample` — empty when every read failed.
    """
    ...

def __format__(self, format_spec: str) -> str:
    """Delegate non-empty format specs to :attr:`value`.

    ``f"{r:.4f}"`` formats the value; ``f"{r}"`` prints the dataclass repr.
    An over-range reading (``value`` is ``None``) formats as ``"None"`` for
    any numeric spec rather than raising.
    """
    if format_spec == "":
        return str(self)
    if self.value is None:
        return "None"
    return format(self.value, format_spec)

def as_dict(self) -> dict[str, float | int | str | bool | None]:
    """Flatten the reading into a row-shaped dict for tabular sinks.

    Content-only — timing provenance lives on the surrounding
    :class:`Sample`. Booleans render as ``0`` / ``1`` so SQLite picks
    INTEGER affinity and CSV / JSONL round-trip cleanly.
    """
    return {
        "channel": self.channel.value,
        "kind": self.kind.value,
        "name": self.name,
        "value": self.value,
        "unit": self.unit.value,
        "fault": int(self.status.fault),
        "maintenance": int(self.status.maintenance),
        "calibrating": int(self.status.calibrating),
        "warming_up": int(self.status.warming_up),
        "alarm1": int(self.status.alarms[0]),
        "alarm2": int(self.status.alarms[1]),
        "alarm3": int(self.status.alarms[2]),
        "alarm4": int(self.status.alarms[3]),
        "ok": int(self.status.ok),
        "protocol": self.protocol.value,
    }

def with_context(self, **updates: Any) -> Self:
    """Return a copy of this error with its context updated.

    Useful when an inner layer raises and an outer layer wants to enrich
    the context (for instance adding ``port`` or ``elapsed_s``).
    """
    cls = type(self)
    new = cls.__new__(cls)
    new.args = self.args
    try:
        new.__dict__.update(self.__dict__)
    except AttributeError:  # pragma: no cover — no slotted subclass today
        for slot in getattr(cls, "__slots__", ()):
            if hasattr(self, slot):
                object.__setattr__(new, slot, getattr(self, slot))
    new.context = self.context.merged(**updates)
    new.__cause__ = self.__cause__
    new.__context__ = self.__context__
    new.__traceback__ = self.__traceback__
    return new

async def add(
    self,
    name: str,
    source: Analyzer | str | Transport,
    *,
    protocol: ProtocolKind = ProtocolKind.MODBUS_RTU,
    address: int = 1,
    serial_settings: SerialSettings | None = None,
    timeout: float = 1.0,
) -> Analyzer:
    """Register an analyser under ``name`` and return it.

    ``source`` discriminates lifecycle ownership: a pre-built :class:`Analyzer`
    (caller-owned, tracked only), a ``str`` port path (the manager opens and
    shares a transport across the bus), or a :class:`Transport` (bound, not
    owned). Continuous-ASCII / ``AUTO`` are refused — the manager is a Modbus
    multidrop coordinator.

    Raises:
        ServomexValidationError: duplicate ``name``, a non-Modbus protocol, or
            a protocol clash with an existing device on the same port.
        ServomexConnectionError: the manager is closed.
    """
    async with self._state_lock:
        self._check_open()
        if name in self._devices:
            raise ServomexValidationError(f"manager: name {name!r} already in use")

        if isinstance(source, Analyzer):
            self._devices[name] = _DeviceEntry(name=name, analyzer=source, port_key=None)
            _logger.info("manager.add device=%s port=prebuilt", name)
            return source

        if protocol not in _MODBUS_KINDS:
            raise ServomexValidationError(
                f"manager: only Modbus protocols can be grouped (got {protocol.value}); "
                "a continuous-ASCII analyser is a single broadcaster, not a multidrop peer",
            )

        port_key, port_entry = await self._resolve_port(
            source, protocol=protocol, serial_settings=serial_settings
        )
        analyzer = self._build_analyzer(
            port_entry, name=name, protocol=protocol, address=address, timeout=timeout
        )
        self._devices[name] = _DeviceEntry(name=name, analyzer=analyzer, port_key=port_key)
        port_entry.refs.add(name)
        _logger.info(
            "manager.add device=%s port=%s protocol=%s address=%s",
            name,
            port_key,
            protocol.value,
            address,
        )
        return analyzer

async def close(self) -> None:
    """Tear down every managed analyser and port (LIFO)."""
    await self._close(suppress_errors=False)

async def execute_each[T](
    self,
    op: Callable[[Analyzer], Awaitable[T]],
    names: Sequence[str] | None = None,
) -> dict[str, DeviceResult[T]]:
    """Run ``op(analyzer)`` on every (or named) analyser concurrently across ports."""
    groups = self._group_by_port(self._resolve_names(names))
    results: dict[str, DeviceResult[T]] = {}
    errors: list[ServomexError] = []
    result_lock = anyio.Lock()

    async def _run_group(port_key: str, members: list[str]) -> None:
        lock = self._lock_for(port_key)
        async with lock:
            for member in members:
                try:
                    value = await op(self._devices[member].analyzer)
                except ServomexError as err:
                    async with result_lock:
                        results[member] = DeviceResult.failure(err)
                        errors.append(err)
                else:
                    async with result_lock:
                        results[member] = DeviceResult[T].success(value)

    async with anyio.create_task_group() as tg:
        for port_key, members in groups.items():
            _ = tg.start_soon(_run_group, port_key, members)

    if self._error_policy is ErrorPolicy.RAISE and errors:
        raise ExceptionGroup("manager.execute_each: one or more analysers failed", errors)
    return results

def get(self, name: str) -> Analyzer:
    """Return the analyser registered under ``name``."""
    try:
        return self._devices[name].analyzer
    except KeyError:
        raise ServomexValidationError(f"manager: no analyser named {name!r}") from None

async def poll(
    self, names: Sequence[str] | None = None, *, timeout: float | None = None
) -> Mapping[str, DeviceResult[Frame]]:
    """Read one :class:`Frame` per (or named) analyser, keyed by name.

    Cross-port concurrent, same-port serialised. Always returns a complete
    mapping; per-device failures land in :attr:`DeviceResult.error`.
    """
    groups = self._group_by_port(self._resolve_names(names))
    results: dict[str, DeviceResult[Frame]] = {}
    result_lock = anyio.Lock()

    async def _run_group(port_key: str, members: list[str]) -> None:
        lock = self._lock_for(port_key)
        async with lock:
            for member in members:
                try:
                    frame = await self._devices[member].analyzer.poll(timeout=timeout)
                except ServomexError as err:
                    async with result_lock:
                        results[member] = DeviceResult.failure(err)
                else:
                    async with result_lock:
                        results[member] = DeviceResult[Frame].success(frame)

    async with anyio.create_task_group() as tg:
        for port_key, members in groups.items():
            _ = tg.start_soon(_run_group, port_key, members)
    return results

async def poll_samples(
    self, *, names: Sequence[str] | None = None, timeout: float | None = None
) -> list[Sample]:
    """Poll every (or named) analyser concurrently across ports → flat samples.

    One :class:`Sample` per channel read. Failed devices are dropped and logged
    at WARN (the recorder never sees them). Same-port devices serialise on the
    shared port lock, acquired once per port-group so a coherent snapshot is
    not interleaved. Satisfies :class:`PollSource`.
    """
    groups = self._group_by_port(self._resolve_names(names))
    result_lock = anyio.Lock()
    all_samples: list[Sample] = []

    async def _run_group(port_key: str, members: list[str]) -> None:
        local: list[Sample] = []
        lock = self._lock_for(port_key)
        async with lock:
            for member in members:
                try:
                    local.extend(
                        await self._devices[member].analyzer.poll_samples(timeout=timeout)
                    )
                except ServomexError as err:
                    _logger.warning("manager.poll_failed device=%s error=%r", member, err)
        async with result_lock:
            all_samples.extend(local)

    async with anyio.create_task_group() as tg:
        for port_key, members in groups.items():
            _ = tg.start_soon(_run_group, port_key, members)
    return all_samples

async def remove(self, name: str) -> None:
    """Unregister ``name``, closing the shared transport on the last ref."""
    async with self._state_lock:
        self._check_open()
        if name not in self._devices:
            raise ServomexValidationError(f"manager: no analyser named {name!r}")
        await self._teardown_device(self._devices.pop(name))
        _logger.info("manager.remove device=%s", name)

async def aclose(self) -> None:
    """Stop the owned producer (if any), unsubscribe, and close the stream."""
    if self._task_group is not None:
        self._task_group.cancel()
        await self._task_group.__aexit__(None, None, None)
        self._task_group = None
    if self._on_close is not None:
        self._on_close()
        self._on_close = None
    await self._receiver.aclose()