sartoriuslib.manager¶

async def add(
    self,
    name: str,
    source: Balance | str | Transport,
    *,
    protocol: ProtocolKind = ProtocolKind.XBPI,
    serial_settings: SerialSettings | None = None,
    timeout: float = 1.0,
    src_sbn: int = 0x01,
    dst_sbn: int = 0x09,
    strict: bool = False,
    identify: bool = True,
) -> Balance:
    """Register and open a balance under ``name``.

    The ``source`` discriminates lifecycle ownership:

    - :class:`Balance` — 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:`SerialTransport`,
      canonicalises the port key, and shares the transport across
      balances on the same bus. Mixing xBPI and SBI sessions on a shared
      physical port is refused; one serial link has one active protocol.
    - :class:`Transport` — duck-typed transport. The manager
      invokes :func:`open_device` against it but does *not* take
      transport ownership.

    Args:
        name: Unique manager-level identifier.
        source: One of the three lifecycle shapes above.
        protocol: Which wire protocol to speak (per
            :func:`sartoriuslib.open_device`). Ignored when
            ``source`` is a pre-built :class:`Balance`.
        serial_settings: Override default serial framing. Only
            honoured when ``source`` is a port-string.
        timeout: Per-call default timeout.
        src_sbn: Host xBPI bus address.
        dst_sbn: Balance xBPI bus address.
        strict: Strict prior gating (see design §6.1).
        identify: Run identify on open and cache :class:`DeviceInfo`.

    Returns:
        The opened :class:`Balance`.

    Raises:
        SartoriusValidationError: ``name`` already exists or an
            invalid combination of kwargs was supplied.
        SartoriusConnectionError: Manager is closed.
    """
    async with self._state_lock:
        self._check_open()
        if name in self._devices:
            raise SartoriusValidationError(
                f"manager: name {name!r} already in use",
                context=ErrorContext(extra={"name": name}),
            )
        if serial_settings is not None and not isinstance(source, str):
            raise SartoriusValidationError(
                "manager.add(serial_settings=...) only applies to string port "
                "sources; pre-built Transport / Balance carry their own settings",
                context=ErrorContext(extra={"name": name}),
            )

        port_key, port_entry, balance = await self._resolve_source(
            source,
            protocol=protocol,
            serial_settings=serial_settings,
            timeout=timeout,
            src_sbn=src_sbn,
            dst_sbn=dst_sbn,
            strict=strict,
            identify=identify,
        )

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

        info = balance.info
        _logger.info(
            "manager.add",
            extra={
                "device_name": name,
                "port_key": port_key,
                "model": info.model if info is not None else None,
                "protocol": balance.session.active_protocol.value,
            },
        )
        return balance

async def close(self) -> None:
    """Tear down every managed balance and port (LIFO)."""
    async with self._state_lock:
        if self._closed:
            return
        for name in reversed(list(self._devices.keys())):
            entry = self._devices.pop(name)
            try:
                await self._teardown_device(entry)
            except Exception as err:
                _logger.warning(
                    "manager.close_device_failed",
                    extra={"device_name": name, "error": repr(err)},
                )
        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 balances participate and
    what arguments each gets — supporting the common case of
    "same command, different argument per balance".
    """
    for name in requests_by_name:
        if name not in self._devices:
            raise SartoriusValidationError(
                f"manager.execute: no balance named {name!r}",
                context=ErrorContext(command_name=command.name, extra={"name": name}),
            )
    targets = tuple(requests_by_name.keys())
    name_by_balance_id = {id(entry.balance): entry.name for entry in self._devices.values()}

    async def _execute(balance: Balance) -> Resp:
        return await balance.session.execute(
            command,
            requests_by_name[name_by_balance_id[id(balance)]],
        )

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

def get(self, name: str) -> Balance:
    """Return the balance registered under ``name``."""
    try:
        return self._devices[name].balance
    except KeyError:
        raise SartoriusValidationError(
            f"manager: no balance 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) balance concurrently across ports.

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

    async def _poll(balance: Balance) -> Reading:
        return await balance.poll()

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

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

    If ``name`` was the last balance on a shared port, the
    transport for that port is closed too. A pre-built
    :class:`Balance` 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 SartoriusValidationError(
                f"manager: no balance 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})