alicatlib¶

def replace(self, **updates: Any) -> Self:
    """Return a copy of this config with ``updates`` applied."""
    return replace(self, **updates)

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``).

    Allocates a fresh instance via ``cls.__new__`` and copies attribute
    state directly. Avoids re-invoking ``__init__`` — many subclasses
    (``AlicatMediumMismatchError``, ``AlicatFirmwareError``,
    ``UnknownGasError`` and friends) have bespoke keyword-only
    signatures that don't accept ``(message, *, context=)``, and
    ``copy.copy`` would silently dispatch through them via
    :meth:`Exception.__reduce__`.
    """
    cls = type(self)
    new = cls.__new__(cls)
    # ``Exception`` slot state lives in ``self.args``; reuse it so
    # ``str(err)`` keeps the original message.
    new.args = self.args
    # Copy subclass-specific attributes (value, field_name, required_min, ...).
    # Use ``__dict__`` directly when present (most subclasses), and fall back
    # to slot iteration if a frozen variant ever appears.
    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 __aenter__(self) -> Self:
    """Enter the async context — returns ``self`` for chaining."""
    return self

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    tb: TracebackType | None,
) -> None:
    """Close every managed device + port on exit."""
    del exc_type, exc, tb
    await self.close()

async def add(
    self,
    name: str,
    source: Device | str | Transport | AlicatProtocolClient,
    *,
    unit_id: str = "A",
    serial: SerialSettings | None = None,
    timeout: float = 0.5,
) -> Device:
    """Register and open a device under ``name``.

    The ``source`` discriminates lifecycle ownership:

    - ``Device`` — pre-built (via :func:`open_device` outside the
      manager). The manager only tracks the name mapping; it does
      *not* take lifecycle ownership.
    - ``str`` — serial port path (``"/dev/ttyUSB0"``, ``"COM3"``).
      The manager creates a
      :class:`~alicatlib.transport.serial.SerialTransport` and
      :class:`AlicatProtocolClient`, canonicalises the port key,
      and reuses them across multi-device buses (RS-485).
    - :class:`Transport` — duck-typed transport. The manager wraps
      it in a new client but does *not* take transport ownership
      (the caller keeps open/close responsibility).
    - :class:`AlicatProtocolClient` — use as-is; the manager does
      not close it.

    Args:
        name: Unique manager-level identifier. Must not already
            exist on this manager.
        source: One of the four lifecycle shapes above.
        unit_id: Bus-level letter for the device. ``"A"`` is the
            polling default; multiple devices on the same port
            get distinct unit ids.
        serial: :class:`SerialSettings` override. Only honoured
            when ``source`` is a port-string — ignored otherwise
            (pre-built transports carry their own settings).
        timeout: Default command timeout passed through to
            :func:`open_device`.

    Returns:
        The identified :class:`Device` (a :class:`FlowMeter`,
        :class:`FlowController`, etc. subclass).

    Raises:
        AlicatValidationError: ``name`` already exists or
            ``serial`` was supplied with a non-string source.
        AlicatConnectionError: The manager is closed.
    """
    async with self._state_lock:
        self._check_open()
        if name in self._devices:
            raise AlicatValidationError(
                f"manager: name {name!r} already in use",
                context=ErrorContext(extra={"name": name}),
            )
        if serial is not None and not isinstance(source, str):
            raise AlicatValidationError(
                "manager.add(serial=...) only applies to string port sources; "
                "pre-built Transport / AlicatProtocolClient carry their own settings",
                context=ErrorContext(extra={"name": name}),
            )

        port_key, port_entry, owns_device = await self._resolve_source(
            source,
            serial=serial,
            timeout=timeout,
        )

        # ``open_device`` runs identification + probes. If it raises,
        # we must not leave the port's ref count dangling.
        try:
            if owns_device:
                assert port_entry is not None  # noqa: S101 — narrow for mypy
                device = await open_device(port_entry.client, unit_id=unit_id, timeout=timeout)
            else:
                # ``source`` was a pre-built Device.
                assert isinstance(source, Device)  # noqa: S101 — narrow for mypy
                device = source
        except BaseException:
            if port_entry is not None and port_key is not None and name not in port_entry.refs:
                # We created a brand-new port just for this add —
                # unwind it rather than leaking the transport.
                await self._maybe_teardown_port(port_key, port_entry)
            raise

        self._devices[name] = _DeviceEntry(
            name=name,
            device=device,
            port_key=port_key,
            owns_device=owns_device,
        )
        if port_entry is not None:
            port_entry.refs.add(name)

        _logger.info(
            "manager.add",
            extra={
                "device_name": name,
                "port_key": port_key,
                "unit_id": unit_id,
                "model": device.info.model,
                "firmware": str(device.info.firmware),
            },
        )
        return device

async def close(self) -> None:
    """Tear down every managed device and port (LIFO).

    Idempotent: safe to call from both :meth:`__aexit__` and
    explicit user code. Individual close failures are caught and
    logged so one device's shutdown error doesn't strand the
    others.
    """
    async with self._state_lock:
        if self._closed:
            return
        # Unwind in reverse insertion order — LIFO per design §5.13.
        for name in reversed(list(self._devices.keys())):
            entry = self._devices.pop(name)
            try:
                await self._teardown_device(entry)
            except Exception as err:  # noqa: BLE001 — best-effort manager teardown: keep going so other devices still close
                _logger.warning(
                    "manager.close_device_failed",
                    extra={"device_name": name, "error": repr(err)},
                )
        # Any port entries that survived (e.g. because a pre-built
        # client source never got refs torn down) are left alone —
        # the caller owns them.
        self._closed = True

async def execute[Req, Resp](
    self,
    command: Command[Req, Resp],
    requests_by_name: Mapping[str, Req],
) -> Mapping[str, DeviceResult[Resp]]:
    """Dispatch a per-device ``Command`` across the requested names.

    ``requests_by_name`` chooses both which devices participate and
    what arguments each gets — supporting the common case of
    "same command, different setpoint per device".
    """
    for name in requests_by_name:
        if name not in self._devices:
            raise AlicatValidationError(
                f"manager.execute: no device named {name!r}",
                context=ErrorContext(command_name=command.name, extra={"name": name}),
            )
    targets = tuple(requests_by_name.keys())
    name_by_device_id = {id(entry.device): entry.name for entry in self._devices.values()}

    async def _execute(device: Device) -> Resp:
        return await device.session.execute(
            command,
            requests_by_name[name_by_device_id[id(device)]],
        )

    return await self._dispatch(command.name, targets, _execute)

def get(self, name: str) -> Device:
    """Return the device registered under ``name`` (raises if unknown)."""
    try:
        return self._devices[name].device
    except KeyError:
        raise AlicatValidationError(
            f"manager: no device named {name!r}",
            context=ErrorContext(extra={"name": name}),
        ) from None

async def poll(
    self,
    names: Sequence[str] | None = None,
) -> Mapping[str, DeviceResult[Reading]]:
    """Poll every (or named) device concurrently across ports.

    Returns a mapping from device name to :class:`DeviceResult`
    even under :attr:`ErrorPolicy.RAISE` — but under that policy,
    any failed device's error is re-raised as an
    :class:`ExceptionGroup` after all devices have completed.
    """
    targets = self._resolve_names(names)

    async def _poll(device: Device) -> Reading:
        return await device.poll()

    return await self._dispatch("poll", targets, _poll)

async def remove(self, name: str) -> None:
    """Unregister and close the device named ``name``.

    If ``name`` was the last device on a shared port, the
    transport and client for that port are closed too. A
    pre-built :class:`Device` source is only dropped from the
    manager's registry — the caller retains lifecycle ownership.
    """
    async with self._state_lock:
        self._check_open()
        if name not in self._devices:
            raise AlicatValidationError(
                f"manager: no device named {name!r}",
                context=ErrorContext(extra={"name": name}),
            )
        entry = self._devices.pop(name)
        await self._teardown_device(entry)
        _logger.info("manager.remove", extra={"device_name": name})

async def request(
    self,
    statistics: Sequence[Statistic | str],
    names: Sequence[str] | None = None,
    *,
    averaging_ms: int = 1,
) -> Mapping[str, DeviceResult[MeasurementSet]]:
    """Run :meth:`Device.request` across devices concurrently.

    Every targeted device receives the same statistic list and
    averaging window — mirroring the primer's ``DV`` semantics.
    """
    targets = self._resolve_names(names)

    async def _request(device: Device) -> MeasurementSet:
        return await device.request(statistics, averaging_ms=averaging_ms)

    return await self._dispatch("request", targets, _request)

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

@classmethod
def success[ValueT](cls, value: ValueT) -> DeviceResult[ValueT]:
    """Build a successful result wrapping ``value``."""
    del cls
    return DeviceResult(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)

@classmethod
def parse(cls, software: str) -> Self:
    """Parse ``software`` into a :class:`FirmwareVersion`.

    Accepts any of the historical shapes: ``"GP"``, ``"GP-10v05"``,
    ``"1v00"``, ``"7v99"``, ``"10v05"``, ``"10v5"``, ``"10.05"``, or those
    substrings embedded in a longer response.

    GP detection: if the string contains a standalone ``GP`` token, the
    family is :attr:`FirmwareFamily.GP`, regardless of any trailing
    ``Nv<major>v<minor>`` suffix. ``major`` / ``minor`` are ``0`` for GP
    (the Nv suffix, when present, is purely cosmetic on GP hardware).

    Args:
        software: Firmware string as reported by the device.

    Returns:
        The parsed version.

    Raises:
        AlicatParseError: If ``software`` contains neither a ``GP`` token
            nor a recognisable ``<major>v<minor>`` / ``<major>.<minor>`` pair.
    """
    is_gp = _GP_PREFIX_RE.search(software) is not None
    numeric_match = _NUMERIC_RE.search(software)

    if is_gp:
        return cls(family=FirmwareFamily.GP, major=0, minor=0, raw=software)

    if numeric_match is None:
        raise AlicatParseError(
            f"Could not parse firmware from {software!r}",
            field_name="software",
            expected="GP or <major>v<minor>",
            actual=software,
        )

    major = int(numeric_match.group("major"))
    minor = int(numeric_match.group("minor"))
    family = _family_for_major(major)
    return cls(family=family, major=major, minor=minor, raw=software)

async def poll(
    self,
    names: Sequence[str] | None = None,
) -> Mapping[str, DeviceResult[Reading]]:
    """Poll every named device (or all under management) concurrently.

    Must return a mapping keyed by the manager-assigned device name.
    Successful polls carry the :class:`Reading` as ``.value``;
    failed ones carry the :class:`~alicatlib.errors.AlicatError` as
    ``.error`` (per :class:`~alicatlib.manager.ErrorPolicy.RETURN`).
    """
    ...

async def poll(
    self,
    names: Iterable[str] | None = None,
) -> Mapping[str, DeviceResult[Reading]]:
    """Poll the wrapped device and return a single-entry mapping."""
    if names is not None and self._name not in set(names):
        return {}
    try:
        reading = await self._device.poll()
    except AlicatError as err:
        failure: DeviceResult[Reading] = DeviceResult(value=None, error=err)
        return {self._name: failure}
    return {self._name: DeviceResult.success(reading)}

def as_dict(self) -> dict[str, float | str | None]:
    """Flatten to a JSON/CSV-friendly dict.

    Produces ``{field_name: value, "status": "HLD,OPL", "received_at": iso8601}``
    — status codes collapse into a single comma-joined sorted string
    (empty when no codes are active) so downstream schema is stable
    across rows. Callers that need per-code boolean columns should
    wrap this themselves; the library picks the schema-stable form.
    """
    result: dict[str, float | str | None] = dict(self.values)
    result["status"] = ",".join(sorted(code.value for code in self.status))
    result["received_at"] = self.received_at.isoformat()
    return result

@classmethod
def from_parsed(
    cls,
    parsed: ParsedFrame,
    *,
    reading_format: DataFrameFormat,
    received_at: datetime,
    t_mono_ns: int,
) -> Reading:
    """Wrap a :class:`ParsedFrame` with timing captured at read time."""
    return cls(
        unit_id=parsed.unit_id,
        reading_format=reading_format,
        values=parsed.values,
        values_by_statistic=parsed.values_by_statistic,
        status=parsed.status,
        received_at=received_at,
        t_mono_ns=t_mono_ns,
    )

def get_float(self, name: str) -> float | None:
    """Return the float value at ``name``, or ``None`` if absent or non-numeric.

    This is the "forgiving" accessor used when a downstream consumer
    wants a numeric value and accepts absence. Text-valued fields and
    the ``--`` sentinel both yield ``None``; exceptions are never
    raised. Callers that need strict behaviour should index
    :attr:`values` directly.
    """
    value = self.values.get(name)
    return value if isinstance(value, float) else None