watlowlib.manager¶

async def add(
    self,
    name: str,
    source: Controller | str | Transport,
    *,
    protocol: ProtocolKind = ProtocolKind.STDBUS,
    address: int = 1,
    serial_settings: SerialSettings | None = None,
    family: ControllerFamily = ControllerFamily.UNKNOWN,
) -> Controller:
    """Register and open a controller under ``name``.

    The ``source`` discriminates lifecycle ownership:

    - :class:`Controller` — pre-built (via
      :func:`watlowlib.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 transport, canonicalises the port key,
      and shares the transport + client across controllers on the
      same bus. Mixing Std Bus and Modbus on a shared physical
      port is refused; one serial link has one active protocol.
    - :class:`Transport` — duck-typed transport. The manager builds
      a session against it but does *not* take transport ownership.

    Args:
        name: Unique manager-level identifier.
        source: One of the three lifecycle shapes above.
        protocol: Wire protocol (``STDBUS`` or ``MODBUS_RTU``).
            Ignored when ``source`` is a pre-built :class:`Controller`.
            ``AUTO`` is rejected — open the controller via
            :func:`open_device` first and register the resulting
            :class:`Controller`.
        address: Bus address. Std Bus accepts ``1..16``; Modbus RTU
            accepts ``1..247``.
        serial_settings: Override default serial framing. Only
            honoured when ``source`` is a port-string.
        family: Best-known :class:`ControllerFamily`. Defaults to
            :attr:`ControllerFamily.UNKNOWN`; the session uses this
            for capability priors and is updated by ``identify()``
            on the returned controller.

    Returns:
        The opened :class:`Controller`.

    Raises:
        WatlowValidationError: ``name`` already exists or an
            invalid combination of kwargs was supplied.
        WatlowConfigurationError: protocol mismatches an existing
            lock on the same port, or ``protocol=AUTO``.
        WatlowConnectionError: Manager is closed.
    """
    async with self._state_lock:
        self._check_open()
        if name in self._devices:
            raise WatlowValidationError(
                f"manager: name {name!r} already in use",
                context=ErrorContext(address=address),
            )
        if serial_settings is not None and not isinstance(source, str):
            raise WatlowValidationError(
                "manager.add(serial_settings=...) only applies to string port "
                "sources; pre-built Transport / Controller carry their own settings",
            )

        port_key, port_entry, controller = await self._resolve_source(
            source,
            protocol=protocol,
            address=address,
            serial_settings=serial_settings,
            family=family,
        )

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

        _logger.info(
            "manager.add device_name=%s port_key=%s protocol=%s address=%s",
            name,
            port_key,
            controller.session.protocol_kind.value,
            controller.session.address,
        )
        return controller

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

    Per-device teardown errors are collected; if any occurred,
    they are raised after the close completes as an
    :class:`ExceptionGroup`. This makes explicit ``await mgr.close()``
    calls fail loud on resource leaks. The async-CM exit path
    swallows the errors instead so an in-flight exception still
    wins (see :meth:`__aexit__`).
    """
    await self._close(suppress_errors=False)

async def execute_each[T](
    self,
    op: Callable[[Controller], Awaitable[T]],
    names: Sequence[str] | None = None,
) -> dict[str, DeviceResult[T]]:
    """Run ``op(controller)`` on every (or named) controller concurrently.

    General-purpose dispatcher used for cross-device snapshots
    (``identify``, ``read_pid``, etc.) where each controller runs
    the same coroutine and the result is keyed by name. Cross-port
    runs concurrently; same-port serialises on the shared client
    lock.

    Under :attr:`ErrorPolicy.RAISE` the method still returns a
    complete result mapping but re-raises an :class:`ExceptionGroup`
    of every per-device error after the task group joins.
    """
    targets = self._resolve_names(names)
    groups = self._group_by_port(targets)
    results: dict[str, DeviceResult[T]] = {}
    errors: list[WatlowError] = []
    result_lock = anyio.Lock()

    async def _run_group(member_names: list[str]) -> None:
        for member in member_names:
            entry = self._devices[member]
            controller = entry.controller
            protocol = controller.session.protocol_kind
            try:
                value = await op(controller)
            except WatlowError as err:
                async with result_lock:
                    results[member] = DeviceResult(
                        value=None,
                        error=err,
                        protocol=protocol,
                    )
                    errors.append(err)
            else:
                async with result_lock:
                    results[member] = DeviceResult(
                        value=value,
                        error=None,
                        protocol=protocol,
                    )

    async with anyio.create_task_group() as tg:
        for member_names in groups.values():
            tg.start_soon(_run_group, member_names)

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

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

async def poll(
    self,
    parameters: Sequence[str | int],
    *,
    names: Sequence[str] | None = None,
    instances: Sequence[int] = (1,),
) -> list[Sample]:
    """Poll every (or named) controller concurrently across ports.

    Returns a flat list of :class:`Sample` — one per (device,
    parameter, instance) read that succeeded. Failed reads are
    dropped from the list and logged at WARN. Cross-port reads run
    concurrently; same-port reads serialise on the shared client
    lock.

    This satisfies the :class:`watlowlib.streaming.PollSource`
    Protocol so a manager can drive :func:`watlowlib.streaming.record`
    directly.
    """
    targets = self._resolve_names(names)
    groups = self._group_by_port(targets)

    result_lock = anyio.Lock()
    all_samples: list[Sample] = []

    async def _run_group(member_names: list[str]) -> None:
        local: list[Sample] = []
        for member in member_names:
            entry = self._devices[member]
            local.extend(
                await poll_controller(
                    entry.controller,
                    name=member,
                    parameters=parameters,
                    instances=instances,
                ),
            )
        async with result_lock:
            all_samples.extend(local)

    async with anyio.create_task_group() as tg:
        for member_names in groups.values():
            tg.start_soon(_run_group, member_names)

    return all_samples

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

    If ``name`` was the last controller on a shared port, the
    transport for that port is closed too. A pre-built
    :class:`Controller` 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 WatlowValidationError(
                f"manager: no controller named {name!r}",
            )
        entry = self._devices.pop(name)
        await self._teardown_device(entry)
        _logger.info("manager.remove device_name=%s", name)