alicatlib.devices¶

def device_class_for(info: DeviceInfo) -> type[Device]:
    """Return the concrete :class:`Device` subclass for ``info``.

    Routing is prefix-based via :data:`MODEL_RULES`. Every current rule
    resolves kind deterministically, so the facade class is
    ``rule.device_cls_map[rule.kind]``.

    Unknown prefixes and kinds not in the rule's map fall back to the
    generic :class:`Device` — the session's :class:`DeviceKind` /
    :class:`Medium` gates still fire, so the fallback is safe: commands
    that don't list ``UNKNOWN`` in ``device_kinds`` simply refuse to
    dispatch.
    """
    rule = _rule_for_model(info.model)
    if rule is None:
        return Device
    return rule.device_cls_map.get(rule.kind, Device)

async def identify_device(
    client: AlicatProtocolClient,
    unit_id: str = "A",
    *,
    model_hint: str | None = None,
) -> DeviceInfo:
    """Run ``VE`` → (optional) ``??M*`` → classify; return :class:`DeviceInfo`.

    Capabilities are *not* populated here; call :func:`probe_capabilities`
    separately and merge via :func:`dataclasses.replace`. That split
    matches design §5.9 and keeps the two concerns testable in isolation.

    Args:
        client: A wired :class:`AlicatProtocolClient`. Identification
            issues ``VE`` and (conditionally) ``??M*``, both of which
            flow through the client's single-in-flight lock.
        unit_id: Polling unit id (``"A"``..``"Z"``).
        model_hint: Required when ``??M*`` can't be reached (GP family,
            V1_V7, or any device that rejects the command). Ignored when
            ``??M*`` succeeds. Raises :class:`AlicatConfigurationError`
            if identification reaches the fallback branch without a hint.
    """
    # The session's firmware gating references self._info.firmware, so
    # to drive VE we need a throwaway DeviceInfo first. We populate the
    # rest after parsing VE.
    ve_session = _bootstrap_session(client, unit_id)
    try:
        ve_result = await ve_session.execute(VE_QUERY, VeRequest())
    except AlicatTimeoutError:
        # GP firmware does not implement VE (design §16.6.8, confirmed on
        # a GP07R100 capture 2026-04-17). Fall through to a VE-less
        # identification path that drives ``??M*`` directly and
        # synthesises a GP :class:`FirmwareVersion` from the M8
        # ``Software`` field. If ``??M*`` also fails, the caller must
        # supply ``model_hint`` and we build a minimal GP DeviceInfo.
        return await _identify_without_ve(client, unit_id, model_hint=model_hint)
    firmware = ve_result.firmware
    firmware_date = ve_result.firmware_date

    # Try ??M* on any family the reachability gate allows. The device
    # may still reject (older numeric firmware that was never updated, or
    # a custom revision); on rejection / parse error we fall through to
    # the model_hint path. Design §16.6 — the primer's 8v28 floor was
    # observed wrong on real 8v17 hardware.
    if _manufacturing_info_reachable(firmware):
        mfg_session = _bootstrap_session(
            client,
            unit_id,
            firmware_override=firmware,
        )
        mfg_info: ManufacturingInfo | None
        try:
            mfg_info = await mfg_session.execute(
                MANUFACTURING_INFO,
                ManufacturingInfoRequest(),
            )
        except (
            AlicatCommandRejectedError,
            AlicatTimeoutError,
            AlicatParseError,
        ):
            # Device doesn't actually support ??M* despite the family
            # check. Fall through to the model_hint path below.
            mfg_info = None
        if mfg_info is not None:
            named = _extract_named_fields(mfg_info)
            model = named["model"]
            if model is None:
                if model_hint is None:
                    raise AlicatConfigurationError(
                        f"??M* returned no model field (M04) and no model_hint was supplied; "
                        f"raw by_code={dict(mfg_info.by_code)!r}",
                        context=ErrorContext(
                            command_name="??M*",
                            unit_id=unit_id,
                            raw_response=None,
                        ),
                    )
                model = model_hint
            return DeviceInfo(
                unit_id=unit_id,
                manufacturer=named["manufacturer"],
                model=model,
                serial=named["serial"],
                manufactured=named["manufactured"],
                calibrated=named["calibrated"],
                calibrated_by=named["calibrated_by"],
                software=named["software"] or str(firmware),
                firmware=firmware,
                firmware_date=firmware_date,
                kind=_kind_for_model(model),
                media=_media_for_model(model),
                capabilities=Capability.NONE,
            )

    # Fallback: ??M* unsupported / rejected, or family says skip.
    if model_hint is None:
        raise AlicatConfigurationError(
            f"Firmware {firmware} does not support ??M*; supply model_hint=... "
            f"to open_device / identify_device to synthesise DeviceInfo "
            f"(GP family / pre-8v28 devices reach this branch — see design §5.9).",
            context=ErrorContext(
                command_name="identify_device",
                unit_id=unit_id,
                firmware=firmware,
            ),
        )
    return DeviceInfo(
        unit_id=unit_id,
        manufacturer=None,
        model=model_hint,
        serial=None,
        manufactured=None,
        calibrated=None,
        calibrated_by=None,
        software=str(firmware),
        firmware=firmware,
        firmware_date=firmware_date,
        kind=_kind_for_model(model_hint),
        media=_media_for_model(model_hint),
        capabilities=Capability.NONE,
    )

@asynccontextmanager
async def open_device(
    port: str | Transport | AlicatProtocolClient,
    *,
    unit_id: str = "A",
    serial: SerialSettings | None = None,
    timeout: float = 0.5,
    recover_from_stream: bool = True,
    model_hint: str | None = None,
    assume_capabilities: Capability = Capability.NONE,
    assume_media: Medium | None = None,
) -> AsyncGenerator[Device]:
    """Open a fully-identified :class:`Device` for ``async with`` use.

    The caller's ``port`` determines the lifecycle the context manager
    takes ownership of:

    - ``str`` (``"/dev/ttyUSB0"`` etc.) — build a
      :class:`SerialTransport` from ``serial`` (or defaults), open it,
      wrap in an :class:`AlicatProtocolClient`, close both on exit.
    - :class:`Transport` — wrap in a new
      :class:`AlicatProtocolClient`; the transport's open/close is the
      caller's responsibility (we never close a transport we didn't
      open).
    - :class:`AlicatProtocolClient` — use as-is; neither transport nor
      client is closed on exit. Stream recovery is skipped because the
      factory doesn't have access to the underlying transport.

    The ``assume_capabilities`` override is union'd onto the probed set
    per design §5.9 — the factory never *subtracts* flags, because
    silently masking hardware the device reports as present is exactly
    the failure mode capability probing exists to avoid.

    The ``assume_media`` override **replaces** the prefix-derived media
    (design §5.9a). Medium answers "how is this specific unit
    configured," not "what can the hardware do" — the common correction
    is to narrow from a permissive prefix default to the single medium
    the unit was actually ordered locked to. The K-family CODA prefixes
    default to ``Medium.GAS | Medium.LIQUID`` because the part-number
    decoder encodes kind but not medium; other future order-configurable
    prefixes can adopt the same pattern. A replace policy also
    future-proofs the model: any new ambiguous prefix drops into
    :data:`MODEL_RULES` with the widest default, and users narrow at
    open time.
    """
    owns_transport = False
    owns_client = False
    transport: Transport | None = None

    if isinstance(port, AlicatProtocolClient):
        client = port
    elif isinstance(port, str):
        settings = serial if serial is not None else SerialSettings(port=port)
        transport = SerialTransport(settings)
        client = AlicatProtocolClient(transport, default_timeout=timeout)
        owns_transport = True
        owns_client = True
    else:
        # Duck-typed Transport (Protocol isn't runtime-checkable).
        transport = port
        client = AlicatProtocolClient(transport, default_timeout=timeout)
        owns_client = True

    try:
        if transport is not None and not transport.is_open:
            await transport.open()
        if recover_from_stream and transport is not None:
            await _recover_from_stream(transport, unit_id)

        info = await identify_device(client, unit_id, model_hint=model_hint)
        probed_caps, probe_report = await probe_capabilities(client, unit_id, info)
        merged_caps = probed_caps | assume_capabilities
        # Medium resolution: prefix-derived by default; ``assume_media``
        # **replaces** (not unions — design §5.9a). Rationale: the common
        # correction is narrowing a permissive prefix default
        # (``Medium.GAS | Medium.LIQUID`` for K-family CODA prefixes and similar
        # whose medium varies by order-time configuration) to the
        # single medium the unit was actually ordered locked to.
        resolved_media = info.media if assume_media is None else assume_media
        info = dataclasses.replace(
            info,
            media=resolved_media,
            capabilities=merged_caps,
            probe_report=probe_report,
        )

        data_frame_format = await _probe_data_frame_format(client, info, unit_id)

        # Per design §10.1: bind per-field engineering units
        # from ``DCU`` where ``??D*`` didn't surface a recognisable label,
        # then populate ``DeviceInfo.full_scale`` from ``FPF`` so
        # setpoint and similar facades can range-check pre-I/O (design
        # §5.20.2). Both probes iterate the data-frame fields, are
        # best-effort per statistic, and never fail the open — a device
        # that rejects one probe just leaves that slot unresolved.
        data_frame_format = await _bind_field_units(
            client,
            info,
            unit_id,
            data_frame_format,
        )
        info = await _probe_full_scales(
            client,
            info,
            unit_id,
            data_frame_format,
        )

        port_label = _resolve_port_label(port, transport)
        session = Session(
            client,
            unit_id=unit_id,
            info=info,
            data_frame_format=data_frame_format,
            port_label=port_label,
        )
        # Pre-cache the loop-control variable for controllers so the
        # first setpoint call can already range-check. Best-effort:
        # firmware without ``LV`` (V1_V7, pre-9v00 V8_V9) leaves the
        # cache ``None`` and setpoint simply skips the range check.
        await _prefetch_loop_control_variable(session, info)

        device_cls = device_class_for(info)
        device = device_cls(session)

        try:
            yield device
        finally:
            await device.close()
    finally:
        del owns_client  # not load-bearing — we never open client separately
        if owns_transport and transport is not None and transport.is_open:
            await transport.close()

async def probe_capabilities(
    client: AlicatProtocolClient,
    unit_id: str,
    info: DeviceInfo,
) -> tuple[Capability, Mapping[Capability, ProbeOutcome]]:
    """Probe each :class:`Capability` flag on the device.

    Real probes implemented so far:

    - :attr:`Capability.BAROMETER` — ``FPF 15`` on any numeric-family
      device. "Present" iff the reply has ``value > 0`` *and*
      ``unit_label != "---"`` (design §16.6.3). The device emits
      ``A <zero> 1 ---`` when the statistic is not supported, which
      has to be disambiguated from a real reading. Note that a
      positive ``BAROMETER`` probe on a flow controller does NOT
      imply :attr:`Capability.TAREABLE_ABSOLUTE_PRESSURE` — the two
      dissociate in practice (design §16.6.7 / Capability docstring).
    - :attr:`Capability.SECONDARY_PRESSURE` — identical rule applied
      to ``FPF 344`` (second absolute pressure). Trying ``344`` covers
      the common second-pressure-sensor configuration; future work can
      extend to ``352`` / ``360`` if devices surface those instead.

    Stubs still fail-closed:

    - :attr:`Capability.TAREABLE_ABSOLUTE_PRESSURE` — no safe probe;
      test-writing ``PC`` would re-zero the abs sensor. Users with a
      pressure meter/controller that supports ``PC`` opt in via
      ``assume_capabilities=Capability.TAREABLE_ABSOLUTE_PRESSURE``.

    - :attr:`Capability.MULTI_VALVE` / :attr:`THIRD_VALVE` —
      ``VD`` returns four columns unconditionally across meter and
      single-valve-controller devices alike (design §16.6.6), so the
      earlier column-count plan is invalidated. Left absent until a
      valve-count signal surfaces.
    - :attr:`Capability.TOTALIZER`, analog I/O flags, display, remote
      tare, bidirectional — no hardware-validated probe strategy yet.

    GP family skips every probe: we have no GP capture and the primer
    doesn't document ``FPF`` there, so assuming absence is the safe
    default. Callers can still union capabilities in via
    ``assume_capabilities=...`` on :func:`open_device`.

    Fails *closed* on every flag the probe doesn't positively confirm —
    design §5.9's "default-absent" policy: a probe that can't answer
    should never falsely claim the hardware is present.
    """
    report: dict[Capability, ProbeOutcome] = dict.fromkeys(
        _iter_single_capability_flags(),
        "absent",
    )

    if info.firmware.family is FirmwareFamily.GP:
        # No FPF on GP; leave every flag absent.
        _logger.info(
            "probe_capabilities.gp_skip",
            extra={
                "unit_id": unit_id,
                "firmware": str(info.firmware),
                "reason": "gp_family_no_fpf",
                "resolved": "NONE",
            },
        )
        return Capability.NONE, MappingProxyType(report)

    probe_session = _bootstrap_session(client, unit_id, firmware_override=info.firmware)

    baro_flag, baro_outcome = await _probe_fpf_capability(
        probe_session,
        statistic=_STAT_BARO_PRESS,
        present_flag=Capability.BAROMETER,
    )
    report[Capability.BAROMETER] = baro_outcome

    sec_flag, sec_outcome = await _probe_fpf_capability(
        probe_session,
        statistic=_STAT_ABS_PRESS_SECOND,
        present_flag=Capability.SECONDARY_PRESSURE,
    )
    report[Capability.SECONDARY_PRESSURE] = sec_outcome

    resolved = baro_flag | sec_flag
    # Structured INFO per design §5.19 / §15.2: one log event per
    # identification summarising the capability-probe outcome. Users
    # wiring up dashboards / dashboards-by-error want to see this as
    # a single row per device rather than scraping per-probe entries.
    _logger.info(
        "probe_capabilities.result",
        extra={
            "unit_id": unit_id,
            "firmware": str(info.firmware),
            "model": info.model,
            "resolved": resolved.name or str(resolved),
            "present": [c.name for c in _iter_single_capability_flags() if report[c] == "present"],
            "outcomes": {c.name: report[c] for c in _iter_single_capability_flags() if c.name},
        },
    )
    return resolved, MappingProxyType(report)

async def find_devices(
    ports: Iterable[str] | None = None,
    *,
    unit_ids: Sequence[str] = ("A",),
    baudrates: Sequence[int] = DEFAULT_DISCOVERY_BAUDRATES,
    timeout: float = _DEFAULT_PROBE_TIMEOUT_S,
    max_concurrency: int = _DEFAULT_MAX_CONCURRENCY,
    stop_on_first_hit: bool = False,
) -> tuple[DiscoveryResult, ...]:
    """Probe the cross-product ``ports × unit_ids × baudrates`` concurrently.

    When ``ports`` is ``None`` the sweep enumerates every port visible
    via :func:`list_serial_ports` — convenient for "what's plugged in?"
    but note that a large fleet plus multiple baudrates multiplies out
    quickly (10 ports × 2 baud × 5 unit ids = 100 probes).

    Concurrency is bounded two ways:

    - ``max_concurrency`` via :class:`anyio.CapacityLimiter` — at most
      that many serial handles are ever open simultaneously.
    - A per-port :class:`anyio.Lock` — combinations targeting the same
      physical port serialise, because a serial port can only be held
      by one transport at a time. Without this, a sweep that tries two
      baud rates on one port would see the second probe fail with
      ``PortBusyError`` (or an unrelated transport error) even when the
      device is present at the correct baud — the two probes simply
      raced for the same handle.

    Lock order is port-first, limiter-second: a probe waiting on its
    port lock does not consume a limiter slot, which keeps the overall
    concurrency ceiling meaningful.

    When ``stop_on_first_hit`` is ``True``, a successful probe at
    ``(port, _, baud)`` records ``baud`` as that port's confirmed rate
    and any pending same-port probe at a different baud is skipped.
    Same-baud probes at other unit ids still run (important for RS-485
    multi-drop buses where several devices share a port at a single
    baud). Skipped combinations are simply omitted from the result
    tuple, so the caller can expect ``len(result) ≤ len(combinations)``.
    Default is ``False`` — every combination produces a result, in a
    stable row-major order (``ports`` × ``unit_ids`` × ``baudrates``).

    The function never raises — every probe's result lands in the
    returned tuple, ``ok`` or not.
    """
    if ports is None:
        ports = await list_serial_ports()
    port_list = list(ports)

    combinations = list(product(port_list, unit_ids, baudrates))
    results: list[DiscoveryResult | None] = [None] * len(combinations)
    limiter = anyio.CapacityLimiter(max_concurrency)
    port_locks: dict[str, anyio.Lock] = {port: anyio.Lock() for port in port_list}
    # Per-port confirmed baud — populated on first ok result under
    # ``stop_on_first_hit``. Keyed by port because baud is a bus
    # property, not a per-device one: if one unit id responded at
    # 19200, the bus is at 19200 and other bauds are pointless.
    confirmed_baud: dict[str, int] = {}

    async def _run(index: int, port: str, unit_id: str, baudrate: int) -> None:
        async with port_locks[port]:
            if stop_on_first_hit:
                hit = confirmed_baud.get(port)
                if hit is not None and hit != baudrate:
                    return
            async with limiter:
                result = await probe(
                    port,
                    unit_id=unit_id,
                    baudrate=baudrate,
                    timeout=timeout,
                )
            results[index] = result
            if stop_on_first_hit and result.ok:
                confirmed_baud[port] = baudrate

    async with anyio.create_task_group() as tg:
        for index, (port, unit_id, baudrate) in enumerate(combinations):
            tg.start_soon(_run, index, port, unit_id, baudrate)

    # ``None`` entries are skipped-by-design under ``stop_on_first_hit``;
    # otherwise every slot is populated because the task group only
    # exits after every spawned task returns.
    return tuple(r for r in results if r is not None)

async def list_serial_ports() -> list[str]:
    """Enumerate serial-port device paths visible to the OS.

    Thin wrapper over :func:`anyserial.list_serial_ports`. Returns
    device-path strings (``/dev/ttyUSB0``, ``COM3`` …) in whatever order
    the backend reports.

    The native backend does not require the ``anyserial[discovery-pyserial]``
    extra; platforms where it misses devices can install that extra and
    switch by setting the ``backend="pyserial"`` kwarg on
    :func:`anyserial.list_serial_ports` directly.
    """
    return [port.device for port in await anyserial.list_serial_ports()]

async def probe(
    port: str,
    *,
    unit_id: str = "A",
    baudrate: int = 19200,
    timeout: float = _DEFAULT_PROBE_TIMEOUT_S,
) -> DiscoveryResult:
    """Probe one port at one baudrate for one unit id.

    Never raises — every failure becomes :attr:`DiscoveryResult.error`
    so that a bulk :func:`find_devices` call collects a uniform result
    set. Opening errors (permission denied, port busy, no such device)
    are caught here the same as identification errors; the caller sees
    one shape whether the device is offline, misconfigured, or silent.
    """
    settings = SerialSettings(port=port, baudrate=baudrate)
    transport = SerialTransport(settings)
    try:
        await transport.open()
    except AlicatError as err:
        return DiscoveryResult(
            port=port,
            unit_id=unit_id,
            baudrate=baudrate,
            info=None,
            error=err,
        )
    try:
        client = AlicatProtocolClient(
            transport,
            default_timeout=timeout,
            # Multiline (``??M*``) deserves a bit more headroom — the
            # factory-default ratio of 2x matches the protocol client
            # itself.
            multiline_timeout=timeout * 2,
        )
        return await _probe_with_client(
            client,
            port=port,
            unit_id=unit_id,
            baudrate=baudrate,
        )
    finally:
        # Best-effort teardown — a close failure here shouldn't hide
        # the identification result the caller came for.
        with contextlib.suppress(AlicatError):
            await transport.close()