watlowlib.devices¶

async def aclose(self) -> None:
    """Close the underlying transport and dispose the protocol client."""
    # The session holds a reference to the protocol client; dispose
    # it so any pending caller learns the controller is gone before
    # the transport close races them.
    try:
        self._session.dispose()
    finally:
        await self._transport.close()

async def identify(
    self,
    *,
    timeout: float | None = None,
    strict: bool = False,
    query_configured_protocol: bool = False,
) -> DeviceInfo:
    """Read the identity parameters and return a :class:`DeviceInfo`.

    Reads (in order): part number (1009), hardware id (1001),
    firmware id (1002), serial number. Missing secondary fields
    stay ``None`` and the result's :attr:`DeviceInfo.health` is
    promoted from :attr:`DeviceHealth.OK` to
    :attr:`DeviceHealth.PARTIAL`. If the part-number read itself
    fails, the result's health is :attr:`DeviceHealth.FAILED` and
    capability decoding is skipped (the family prior still
    applies).

    Args:
        timeout: Per-read timeout override.
        strict: If ``True``, raise the underlying error when the
            part-number read fails instead of returning a
            ``health=FAILED`` info. Use this in maintenance code
            paths that need to know the device actually answered
            before declaring success.
        query_configured_protocol: If ``True``, also read parameter
            17009 (Protocol) and populate
            :attr:`DeviceInfo.configured_protocol`. Off by default
            because the read costs an extra round-trip; the
            maintenance verify pass and the discover CLI opt in.

    Raises:
        WatlowError: When ``strict=True`` and the part-number read
            fails. The original transport / protocol error class
            is preserved.
    """
    if strict:
        entry = await self.read_parameter("part_number", timeout=timeout)
        part_raw = entry.value if isinstance(entry.value, str) else None
    else:
        part_raw = await self._safe_read_str("part_number", timeout=timeout)
    hw_id = await self._safe_read_int("hardware_id", timeout=timeout)
    fw_id = await self._safe_read_int("firmware_id", timeout=timeout)
    serial_str = await self._safe_read_str("serial_number", timeout=timeout)

    if part_raw:
        part = decode_part_number(part_raw)
        capabilities = capabilities_for_part_number(part)
        # PARTIAL when part_number is fine but a secondary read missed.
        secondary_missing = hw_id is None or fw_id is None
        health = DeviceHealth.PARTIAL if secondary_missing else DeviceHealth.OK
    else:
        part = PartNumber(raw="", family=ControllerFamily.UNKNOWN)
        # No part number → no SKU decode; capability table degrades
        # to the UNKNOWN family prior (NONE).
        capabilities = capabilities_for_part_number(part)
        health = DeviceHealth.FAILED

    configured_protocol: ProtocolKind | None = None
    if query_configured_protocol:
        code = await self._safe_read_int(17009, timeout=timeout)
        if code is not None:
            configured_protocol = _PROTOCOL_CODE_TO_KIND.get(code)

    loops = default_loops(part)
    # Cache for ``self.loop(n)``'s eager validator. Identify is the
    # canonical place that gets to set this — open() doesn't have
    # the part number yet, and a SKU's loop count / capabilities
    # never change mid-session.
    self._loops = loops
    self._capabilities = capabilities
    return DeviceInfo(
        part_number=part,
        hardware_id=hw_id,
        firmware_id=fw_id,
        serial_number=serial_str,
        family=part.family,
        protocol=self._session.protocol_kind,
        address=self._session.address,
        capabilities=capabilities,
        serial_settings=self._serial_settings,
        loops=loops,
        health=health,
        configured_protocol=configured_protocol,
    )

def loop(self, n: int) -> ControllerLoop:
    """Return a sub-facade bound to loop ``n`` (1-indexed).

    ``n`` is validated eagerly when :attr:`loops` is known,
    otherwise per-spec ``max_instance`` validation kicks in at the
    first wire call. Multi-loop access is the public way to reach
    loop 2 on dual-loop devices —
    :meth:`Controller.read_pv` defaults to ``instance=1``.
    """
    return ControllerLoop(self, n)

async def poll(
    self,
    parameters: Sequence[str | int],
    *,
    names: Sequence[str] | None = None,
    instances: Sequence[int] = (1,),
) -> list[Sample]:
    r"""Read every (parameter × instance) and return them as :class:`Sample`\ s.

    Satisfies the :class:`watlowlib.streaming.PollSource` Protocol so
    a solo :class:`Controller` can drive :func:`watlowlib.streaming.record`
    directly without a manager. ``names`` is accepted for Protocol
    compatibility but ignored — a Controller has only one device.

    Failed reads are dropped from the returned list and logged at
    WARN. The recorder treats absence as "drop this row from the
    batch" and continues with the next tick.
    """
    del names  # solo controller has no name-keyed device map
    from watlowlib.streaming._poll import poll_controller  # noqa: PLC0415 — avoid cycle

    return await poll_controller(
        self,
        name=self._transport.label,
        parameters=parameters,
        instances=instances,
    )

async def read_parameter(
    self,
    name_or_id: str | int,
    *,
    instance: int = 1,
    timeout: float | None = None,
) -> ParameterEntry:
    """Read any registry parameter.

    ``instance=1`` is the default for single-loop devices and the
    first loop / channel on multi-loop devices.
    """
    return await self._session.execute(
        READ_PARAMETER,
        ReadParameterRequest(name_or_id, instance=instance),
        timeout=timeout,
    )

async def read_pv(self, *, instance: int = 1, timeout: float | None = None) -> Reading:
    """Read the process value for ``instance`` (loop number, 1-indexed)."""
    entry = await self.read_parameter("process_value", instance=instance, timeout=timeout)
    return self._reading_from_entry(entry)

async def read_setpoint(self, *, instance: int = 1, timeout: float | None = None) -> Reading:
    """Read the active setpoint for ``instance``."""
    entry = await self.read_parameter("setpoint", instance=instance, timeout=timeout)
    return self._reading_from_entry(entry)

async def set_setpoint(
    self,
    value: float,
    *,
    instance: int = 1,
    confirm: bool = False,
    timeout: float | None = None,
) -> Reading:
    """Write the setpoint and return the device-echoed value as a :class:`Reading`.

    Setpoint is RWES — pass ``confirm=True`` to acknowledge the
    EEPROM write. The returned reading is the device's echo of
    the value it accepted.
    """
    entry = await self.write_parameter(
        "setpoint",
        value,
        instance=instance,
        confirm=confirm,
        timeout=timeout,
    )
    return self._reading_from_entry(entry)

async def write_parameter(
    self,
    name_or_id: str | int,
    value: float | int | str,
    *,
    instance: int = 1,
    confirm: bool = False,
    timeout: float | None = None,
) -> ParameterEntry:
    """Write any registry parameter.

    Persistent (RWE / RWES) writes require ``confirm=True``;
    the session raises :class:`WatlowConfirmationRequiredError`
    before any I/O if the gate is missing.
    """
    return await self._session.execute(
        WRITE_PARAMETER,
        WriteParameterRequest(name_or_id, value, instance=instance),
        confirm=confirm,
        timeout=timeout,
    )

async def read_alarms(self) -> AlarmState:
    """Read the alarm word for this loop.

    Currently raises :class:`watlowlib.errors.WatlowProtocolUnsupportedError` —
    see :func:`watlowlib.commands.alarms.read_alarms` for why the
    decoder is not yet wired up.
    """
    return await _read_alarms(self._controller.session, instance=self._loop)

async def read_output(self) -> Reading:
    """Read this loop's working output (``output_power``)."""
    return await _read_output(self._controller.session, instance=self._loop)

async def read_pid(self) -> PidGains:
    """Read every PID gain for this loop. Missing gains return ``None``.

    Cool-side gains (``cool_proportional_band``, ``dead_band``)
    are skipped when the controller's identified capabilities
    lack :attr:`Capability.HAS_COOLING` (e.g. PM ``output_2 ==
    'A'``). Pre-identify, the gate is permissive.
    """
    return await _read_pid(
        self._controller.session,
        instance=self._loop,
        capabilities=self._controller.capabilities,
    )

async def read_pv(self, *, timeout: float | None = None) -> Reading:
    """Read this loop's process value."""
    return await self._controller.read_pv(instance=self._loop, timeout=timeout)

async def read_setpoint(self, *, timeout: float | None = None) -> Reading:
    """Read this loop's active setpoint."""
    return await self._controller.read_setpoint(instance=self._loop, timeout=timeout)

async def set_setpoint(
    self,
    value: float,
    *,
    confirm: bool = False,
    timeout: float | None = None,
) -> Reading:
    """Write this loop's setpoint (RWES → ``confirm=True`` required)."""
    return await self._controller.set_setpoint(
        value,
        instance=self._loop,
        confirm=confirm,
        timeout=timeout,
    )

async def write_pid(self, gains: PidGains, *, confirm: bool = False) -> PidGains:
    """Write the supplied gains for this loop.

    Persistent — passing ``confirm=True`` is required. Fields
    left ``None`` on ``gains`` skip the wire entirely. Setting a
    cool-side field on a controller without
    :attr:`Capability.HAS_COOLING` raises
    :class:`watlowlib.errors.WatlowConfigurationError`.
    """
    return await _write_pid(
        self._controller.session,
        gains,
        instance=self._loop,
        confirm=confirm,
        capabilities=self._controller.capabilities,
    )

def availability(self, command_name: str) -> Availability:
    """Cached availability for ``command_name``."""
    return self._availability.get(command_name, Availability.UNKNOWN)

def dispose(self) -> None:
    """Dispose the bound protocol client."""
    self._client.dispose()

async def execute[Req, Resp](
    self,
    command: Command[Req, Resp],
    request: Req,
    *,
    confirm: bool = False,
    timeout: float | None = None,
) -> Resp:
    """Dispatch ``command`` with ``request`` and return the typed response."""
    kind = self._client.kind
    # Variant resolution. The session picks the variant matching
    # the bound protocol; one protocol per port (invariant 1).
    # Resolve to a single ``variant`` local so the rest of the
    # method is protocol-agnostic; we still branch on ``kind``
    # for the encode/decode call shapes (stdbus takes ``reply``;
    # modbus takes ``words, ctx, request``).
    if kind is ProtocolKind.STDBUS:
        stdbus_variant = command.stdbus
        if stdbus_variant is None:
            raise WatlowProtocolUnsupportedError(
                f"command {command.name!r} has no Std Bus variant",
                context=self._error_context(command, request),
            )
        modbus_variant = None
    elif kind is ProtocolKind.MODBUS_RTU:
        modbus_variant = command.modbus
        if modbus_variant is None:
            raise WatlowProtocolUnsupportedError(
                f"command {command.name!r} has no Modbus variant",
                context=self._error_context(command, request),
            )
        stdbus_variant = None
    else:
        raise WatlowProtocolUnsupportedError(
            f"session has unsupported protocol kind {kind!r}",
            context=self._error_context(command, request),
        )

    # Cache key. We key on ``command_name:parameter_id`` for
    # registry-driven commands so that one ``read_parameter("foo")``
    # rejection doesn't sticky-block every other parameter; bare
    # commands fall back to ``command.name``.
    cache_key = self._cache_key(command, request)

    cached = self._availability.get(cache_key, Availability.UNKNOWN)
    if cached is Availability.UNSUPPORTED:
        raise WatlowProtocolUnsupportedError(
            f"command {command.name!r} is unsupported on this device",
            context=self._error_context(command, request),
        )

    # Safety gate: PERSISTENT writes need explicit confirm.
    if command.safety is SafetyTier.PERSISTENT and not confirm:
        raise WatlowConfirmationRequiredError(
            f"command {command.name!r} is PERSISTENT and requires confirm=True",
            context=self._error_context(command, request),
        )

    ctx = CommandContext(
        registry=self._registry,
        family=self._family,
        address=self._address,
        port=self._port,
    )

    bound_timeout = timeout if timeout is not None else DEFAULTS.io_timeout_s

    # Encode under the variant. Errors here are pre-I/O — typically
    # validation failures — and should propagate untouched.
    # Exactly one of ``stdbus_variant`` / ``modbus_variant`` is
    # non-None per the resolution above; the type narrowing is
    # explicit so neither mypy nor pyright needs an ``assert`` it
    # can't enforce at runtime under ``-O``.
    wire_request: Any
    if stdbus_variant is not None:
        wire_request = stdbus_variant.encode(ctx, request)
    elif modbus_variant is not None:
        wire_request = modbus_variant.encode(ctx, request)
    else:  # pragma: no cover — variant resolution above guarantees one is set
        raise WatlowProtocolUnsupportedError(
            f"command {command.name!r} variant resolution lost",
            context=self._error_context(command, request),
        )

    started = time.monotonic()
    # Hold the per-port client lock only for the I/O turn-around.
    # Decode is CPU-only and does not need to block the next
    # request waiting on the same RS-485 segment; ``reply`` is
    # snapshotted before the lock releases.
    async with self._client.lock:
        try:
            reply = await self._client.execute(
                wire_request,
                address=self._address,
                timeout=bound_timeout,
                command_name=command.name,
            )
        except (
            WatlowNoSuchObjectError,
            WatlowNoSuchAttributeError,
            WatlowProtocolUnsupportedError,
        ) as exc:
            self._availability[cache_key] = Availability.UNSUPPORTED
            _log.warning(
                "command unsupported: protocol=%s cmd=%s key=%s exc=%s",
                kind.value,
                command.name,
                cache_key,
                exc,
            )
            raise
        except WatlowProtocolError:
            raise
        except WatlowError as exc:
            _log.warning(
                "command error: protocol=%s cmd=%s key=%s exc=%s",
                kind.value,
                command.name,
                cache_key,
                exc,
            )
            raise

    # Decode outside the lock — pure compute on the captured reply.
    try:
        if stdbus_variant is not None:
            response = stdbus_variant.decode(reply, ctx)
        else:
            # ``modbus_variant is not None`` per the resolution above;
            # mypy/pyright follow the narrowing without an ``assert``.
            response = modbus_variant.decode(reply, ctx, request)  # type: ignore[union-attr]
    except (
        WatlowNoSuchObjectError,
        WatlowNoSuchAttributeError,
        WatlowProtocolUnsupportedError,
    ) as exc:
        # Decode-side "we don't have this": same availability
        # transition as the wire-side rejection above.
        self._availability[cache_key] = Availability.UNSUPPORTED
        _log.warning(
            "command unsupported: protocol=%s cmd=%s key=%s exc=%s",
            kind.value,
            command.name,
            cache_key,
            exc,
        )
        raise
    except WatlowProtocolError:
        # Decode-failure parity with the inside-lock branch above:
        # NoSuchInstance / IllegalDataValue / generic decode errors
        # don't transition availability per design §5b.
        raise

    elapsed = time.monotonic() - started
    self._availability[cache_key] = Availability.SUPPORTED
    _log.debug(
        "session exec ok protocol=%s cmd=%s key=%s elapsed=%.4fs",
        kind.value,
        command.name,
        cache_key,
        elapsed,
    )
    return response

async def aclose(self) -> None:
    """Close the underlying transport and dispose the protocol client."""
    # The session holds a reference to the protocol client; dispose
    # it so any pending caller learns the controller is gone before
    # the transport close races them.
    try:
        self._session.dispose()
    finally:
        await self._transport.close()

async def identify(
    self,
    *,
    timeout: float | None = None,
    strict: bool = False,
    query_configured_protocol: bool = False,
) -> DeviceInfo:
    """Read the identity parameters and return a :class:`DeviceInfo`.

    Reads (in order): part number (1009), hardware id (1001),
    firmware id (1002), serial number. Missing secondary fields
    stay ``None`` and the result's :attr:`DeviceInfo.health` is
    promoted from :attr:`DeviceHealth.OK` to
    :attr:`DeviceHealth.PARTIAL`. If the part-number read itself
    fails, the result's health is :attr:`DeviceHealth.FAILED` and
    capability decoding is skipped (the family prior still
    applies).

    Args:
        timeout: Per-read timeout override.
        strict: If ``True``, raise the underlying error when the
            part-number read fails instead of returning a
            ``health=FAILED`` info. Use this in maintenance code
            paths that need to know the device actually answered
            before declaring success.
        query_configured_protocol: If ``True``, also read parameter
            17009 (Protocol) and populate
            :attr:`DeviceInfo.configured_protocol`. Off by default
            because the read costs an extra round-trip; the
            maintenance verify pass and the discover CLI opt in.

    Raises:
        WatlowError: When ``strict=True`` and the part-number read
            fails. The original transport / protocol error class
            is preserved.
    """
    if strict:
        entry = await self.read_parameter("part_number", timeout=timeout)
        part_raw = entry.value if isinstance(entry.value, str) else None
    else:
        part_raw = await self._safe_read_str("part_number", timeout=timeout)
    hw_id = await self._safe_read_int("hardware_id", timeout=timeout)
    fw_id = await self._safe_read_int("firmware_id", timeout=timeout)
    serial_str = await self._safe_read_str("serial_number", timeout=timeout)

    if part_raw:
        part = decode_part_number(part_raw)
        capabilities = capabilities_for_part_number(part)
        # PARTIAL when part_number is fine but a secondary read missed.
        secondary_missing = hw_id is None or fw_id is None
        health = DeviceHealth.PARTIAL if secondary_missing else DeviceHealth.OK
    else:
        part = PartNumber(raw="", family=ControllerFamily.UNKNOWN)
        # No part number → no SKU decode; capability table degrades
        # to the UNKNOWN family prior (NONE).
        capabilities = capabilities_for_part_number(part)
        health = DeviceHealth.FAILED

    configured_protocol: ProtocolKind | None = None
    if query_configured_protocol:
        code = await self._safe_read_int(17009, timeout=timeout)
        if code is not None:
            configured_protocol = _PROTOCOL_CODE_TO_KIND.get(code)

    loops = default_loops(part)
    # Cache for ``self.loop(n)``'s eager validator. Identify is the
    # canonical place that gets to set this — open() doesn't have
    # the part number yet, and a SKU's loop count / capabilities
    # never change mid-session.
    self._loops = loops
    self._capabilities = capabilities
    return DeviceInfo(
        part_number=part,
        hardware_id=hw_id,
        firmware_id=fw_id,
        serial_number=serial_str,
        family=part.family,
        protocol=self._session.protocol_kind,
        address=self._session.address,
        capabilities=capabilities,
        serial_settings=self._serial_settings,
        loops=loops,
        health=health,
        configured_protocol=configured_protocol,
    )

def loop(self, n: int) -> ControllerLoop:
    """Return a sub-facade bound to loop ``n`` (1-indexed).

    ``n`` is validated eagerly when :attr:`loops` is known,
    otherwise per-spec ``max_instance`` validation kicks in at the
    first wire call. Multi-loop access is the public way to reach
    loop 2 on dual-loop devices —
    :meth:`Controller.read_pv` defaults to ``instance=1``.
    """
    return ControllerLoop(self, n)

async def poll(
    self,
    parameters: Sequence[str | int],
    *,
    names: Sequence[str] | None = None,
    instances: Sequence[int] = (1,),
) -> list[Sample]:
    r"""Read every (parameter × instance) and return them as :class:`Sample`\ s.

    Satisfies the :class:`watlowlib.streaming.PollSource` Protocol so
    a solo :class:`Controller` can drive :func:`watlowlib.streaming.record`
    directly without a manager. ``names`` is accepted for Protocol
    compatibility but ignored — a Controller has only one device.

    Failed reads are dropped from the returned list and logged at
    WARN. The recorder treats absence as "drop this row from the
    batch" and continues with the next tick.
    """
    del names  # solo controller has no name-keyed device map
    from watlowlib.streaming._poll import poll_controller  # noqa: PLC0415 — avoid cycle

    return await poll_controller(
        self,
        name=self._transport.label,
        parameters=parameters,
        instances=instances,
    )