alicatlib.manager¶

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, device_ctx = await self._resolve_source(
            source,
            unit_id=unit_id,
            serial=serial,
            timeout=timeout,
        )

        # ``open_device`` context-enter runs identification + probes.
        # If it raises, we must not leave the port's ref count dangling.
        try:
            if device_ctx is not None:
                device = await device_ctx.__aenter__()
            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,
            device_ctx=device_ctx,
        )
        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[DataFrame]]:
    """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) -> DataFrame:
        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)