watlowlib.devices¶

async def close(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.
    """
    # Device-neutral: the bound profile owns the family-specific
    # identity sequence (EZ-ZONE PM reads 1009/1001/1002/serial;
    # Series SD reads the numeric 10/11/13 + serial 7-8 + reg 18).
    info = await self._session.profile.identify(
        self,
        timeout=timeout,
        strict=strict,
        query_configured_protocol=query_configured_protocol,
    )
    # Cache for ``self.loop(n)``'s eager validator and ``snapshot``.
    # Identify is the canonical place that sets these — open()
    # doesn't have the identity yet, and a device's loop count /
    # capabilities never change mid-session.
    self._loops = info.loops
    self._capabilities = info.capabilities
    self._device_info = info
    return info

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, *, instance: int = 1, timeout: float | None = None) -> Reading:
    """Read the active process value — the canonical no-arg snapshot.

    Equivalent to :meth:`read_pv`. The no-arg form aligns with the
    ecosystem ``poll()`` convention shared by ``alicatlib.Device``,
    ``sartoriuslib.Balance``, and ``nidaqlib.DaqSession``: a
    single, default-shaped reading per call.

    For multi-parameter polling use :meth:`poll_many`.
    """
    return await self.read_pv(instance=instance, timeout=timeout)

async def poll_many(
    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_comms_unit_label(self, *, timeout: float | None = None) -> Unit | None:
    """Read (and cache) the value parameter 17050 reports.

    Inspection / diagnostics helper. **Does not** drive
    :class:`Reading.unit`: on at least one PM3 firmware revision
    17050 is a label-only register that changes the enum the
    device reports for itself but does not affect the scale of
    temperature values exchanged over comms.

    To tell watlowlib what scale temperatures actually travel in
    over the wire, pass ``assert_wire_temperature_unit=`` to
    :func:`watlowlib.open_device`. That assertion is what feeds
    :class:`Reading.unit`.

    Distinct from ``read_parameter("units")``, which targets
    parameter 3005 (front-panel display). The two can disagree on
    a real device.

    Returns ``None`` if the device doesn't report a known code.
    """
    del timeout  # comms_unit_label() is cached + uses session defaults
    return await self._session.comms_unit_label()

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 await reading_from_entry(self._session, 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 await reading_from_entry(self._session, entry)

async def set_comms_unit_label(
    self,
    unit: Unit | str,
    *,
    confirm: bool = False,
    timeout: float | None = None,
) -> Unit | None:
    """Set parameter 17050 ("Communications - Display Units").

    Accepts a :class:`Unit` or a case-insensitive string alias
    (``"C"`` / ``"F"`` / ``"celsius"`` / ``"fahrenheit"`` /
    ``"degC"`` / ``"degF"`` / ``"°C"`` / ``"°F"``).
    :attr:`Unit.PERCENT` is rejected pre-I/O — the register is
    temperature-only.

    Raw enumeration codes (15 / 30) are not accepted here. Callers
    who want the lower-level path use
    ``write_parameter("display_units", 30)``.

    .. warning::

        On at least one PM3 firmware revision this register is
        **label-only**: writing it changes the enum the device
        reports when 17050 is read back, but does not change the
        scale of temperature values exchanged over comms. This
        setter therefore does **not** affect
        :class:`Reading.unit`. To tell watlowlib what scale
        temperatures are actually on, pass
        ``assert_wire_temperature_unit=`` to
        :func:`watlowlib.open_device`.

    Persistent write (parameter 17050 is RWE); pass ``confirm=True``
    to acknowledge the EEPROM write. The session raises
    :class:`WatlowConfirmationRequiredError` pre-I/O if missing.

    Returns the device-echoed label after the write. ``None`` if
    the device's echo decodes outside the known codes.
    """
    resolved = coerce_unit(unit)
    code = display_code_for_unit(resolved)
    if code is None:
        raise WatlowValidationError(
            "set_comms_unit_label accepts CELSIUS / FAHRENHEIT only; "
            "PERCENT is not a valid display-unit code",
        )
    # PERSISTENT write — session enforces ``confirm=True`` pre-I/O.
    await self.write_parameter(
        "display_units",
        code,
        confirm=confirm,
        timeout=timeout,
    )
    self._session.invalidate_comms_unit_label()
    return await self._session.comms_unit_label()

async def set_persistent_writes(
    self,
    enabled: bool,
    *,
    confirm: bool = False,
    timeout: float | None = None,
) -> None:
    """Toggle whether subsequent writes persist to non-volatile memory.

    Series-SD-specific. The SD persists every register write to
    EEPROM by default, so a high-rate writer (ramping setpoints, a
    tuning loop) can wear the EEPROM out and brick the controller.
    Writing ``0`` to register 17 keeps subsequent writes in RAM
    only; the device resets register 17 to ``1`` on every power
    cycle, so call ``set_persistent_writes(False)`` once after each
    power-up before a burst of writes (see ``sd_manual.txt`` p.84).

    Args:
        enabled: ``True`` → persist writes to EEPROM (the power-on
            default); ``False`` → keep writes in RAM only (lost on
            power cycle, but spares the EEPROM).
        confirm: The write itself is gated like any other parameter
            write — pass ``confirm=True`` to acknowledge it.
        timeout: Per-write timeout override.

    Raises:
        WatlowConfirmationRequiredError: ``confirm`` is ``False``.
        WatlowValidationError: the bound profile's registry has no
            ``eeprom_write_enable`` parameter (e.g. an EZ-ZONE PM,
            which has no such register).
    """
    await self.write_parameter(
        "eeprom_write_enable",
        1 if enabled else 0,
        confirm=confirm,
        timeout=timeout,
    )

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 await reading_from_entry(self._session, entry)

async def snapshot(self, *, name: str | None = None) -> WatlowDeviceSnapshot:
    """Return an I/O-free :class:`WatlowDeviceSnapshot`.

    Built from cached identity (populated by :meth:`identify`,
    which :func:`watlowlib.open_device` calls by default) plus
    the session's last error and per-command availability cache.
    Does **not** issue any reads — safe to call from monitoring
    loops at high cadence.

    Args:
        name: Override the snapshot's ``name`` field. Defaults to
            the controller's transport label, matching the
            manager-assigned name surfaced into emitted samples.
    """
    info = self._device_info
    model = info.part_number.raw if info is not None else None
    firmware = (
        str(info.firmware_id) if info is not None and info.firmware_id is not None else None
    )
    serial = info.serial_number if info is not None else None
    # Snapshot is built from cached state; availability_summary is
    # a frozen view of the session's UNSUPPORTED-marked commands.
    availability = {
        key: state
        for key, state in self._session.availability_summary().items()
        if state.name == "UNSUPPORTED"
    }
    return WatlowDeviceSnapshot(
        name=name if name is not None else self._transport.label,
        model=model,
        firmware=firmware,
        serial=serial,
        connected=self._transport.is_open,
        last_error=self._session.last_error,
        recoverable_error_count=self._session.recoverable_error_count,
        captured_at=datetime.now(UTC),
        family=info.family if info is not None else None,
        capabilities=self._capabilities if self._capabilities is not None else Capability.NONE,
        availability_summary=availability,
    )

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,
    )

async def __call__(
    self,
    controller: Controller,
    *,
    timeout: float | None = None,
    strict: bool = False,
    query_configured_protocol: bool = False,
) -> DeviceInfo:
    """Return identity information for ``controller``."""
    ...

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

def availability_summary(self) -> Mapping[str, Availability]:
    """Return the current command availability cache.

    Includes every command the session has dispatched; the
    snapshot path filters to the UNSUPPORTED entries.
    """
    return MappingProxyType(dict(self._availability))

def clear_last_error(self) -> None:
    """Reset :attr:`last_error` to ``None`` (manual diagnostics aid)."""
    self._last_error = None

async def comms_unit_label(self) -> Unit | None:
    """Return the value parameter 17050 reports for this session.

    Inspection helper, not a source of truth: on at least one PM3
    firmware revision 17050 is a label-only register that does not
    govern the wire scale. :class:`Reading.unit` is sourced from
    :meth:`wire_temperature_unit` instead.

    Reads the parameter on first call and caches the result for
    the lifetime of the session. Returns ``None`` when the device
    rejects the read or reports an unknown code; the cache
    distinguishes "haven't asked yet" from "asked and got nothing"
    so a rejection does not cost another wire turn-around.

    Invalidated by :meth:`invalidate_comms_unit_label` (called by
    :meth:`Controller.set_comms_unit_label` after a successful
    write).
    """
    if self._comms_unit_label_loaded:
        return self._comms_unit_label
    self._comms_unit_label = await self._fetch_comms_unit_label()
    self._comms_unit_label_loaded = True
    return self._comms_unit_label

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. ``maybe_acquire``
    # reuses an existing acquisition when the caller (e.g.
    # ``poll_controller``'s tick batch, or
    # ``WatlowManager._run_group``'s port batch) already holds
    # the lock — avoids the FIFO-queue starvation that would
    # otherwise let unrelated writers interleave between a
    # batch's reads.
    async with maybe_acquire(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
            self._record_error(exc)
            _log.warning(
                "command unsupported: protocol=%s cmd=%s key=%s exc=%s",
                kind.value,
                command.name,
                cache_key,
                exc,
            )
            raise
        except WatlowProtocolError as exc:
            self._record_error(exc)
            raise
        except WatlowError as exc:
            self._record_error(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
        self._record_error(exc)
        _log.warning(
            "command unsupported: protocol=%s cmd=%s key=%s exc=%s",
            kind.value,
            command.name,
            cache_key,
            exc,
        )
        raise
    except WatlowProtocolError as exc:
        # Decode-failure parity with the inside-lock branch above:
        # NoSuchInstance / IllegalDataValue / generic decode errors
        # don't transition availability per design §5b.
        self._record_error(exc)
        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

def invalidate_comms_unit_label(self) -> None:
    """Drop the cached 17050 value so the next read re-queries the device."""
    self._comms_unit_label = None
    self._comms_unit_label_loaded = False

def set_wire_temperature_unit(self, unit: Unit | None) -> None:
    """Set the wire temperature scale from an authoritative source.

    Called by an identity strategy that has read the device's own
    unit register (e.g. the Series SD reg 18). Unlike the PM path —
    where the unit register can lie, so the value stays a user
    assertion — this is the device telling us its comms scale
    directly, so it is honest to adopt it. Resets the one-shot
    "trusting user-asserted unit" warning since this value is
    device-sourced, not user-asserted.
    """
    self._wire_temperature_unit = unit
    self._wire_temperature_unit_warned = True

def wire_temperature_unit(self) -> Unit | None:
    """Return the user-asserted scale of temperature values on the wire.

    This is what :class:`Reading.unit` / :class:`Sample.unit` get
    tagged with for temperature parameters. ``None`` when the user
    did not pass ``assert_wire_temperature_unit=`` to
    :func:`watlowlib.open_device`; in that case temperature
    readings carry ``unit=None`` rather than guess.

    Pure accessor — no I/O. Logs a one-shot WARN the first time
    an asserted value is consumed so the user-assertion shows up
    plainly in capture logs.
    """
    if self._wire_temperature_unit is not None and not self._wire_temperature_unit_warned:
        _log.warning(
            "trusting user-asserted wire temperature unit %s for port=%s "
            "address=%s; not independently verified by the library "
            "(parameter 17050 is label-only on at least one PM firmware)",
            self._wire_temperature_unit.value,
            self._port,
            self._address,
        )
        self._wire_temperature_unit_warned = True
    return self._wire_temperature_unit