watlowlib¶

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 close(self) -> None:
    """Flush and close the CSV file. Idempotent."""
    if self._file is None:
        return
    try:
        self._file.flush()
    finally:
        self._file.close()
        self._file = None
        self._writer = None

async def open(self) -> None:
    """Open the CSV file for writing.

    Truncates any existing file: the first :meth:`write_many` will
    write a fresh header row. Idempotent on already-open sinks.
    """
    if self._file is not None:
        return
    self._file = self._path.open("w", encoding="utf-8", newline="")

async def write_many(self, samples: Sequence[Sample]) -> None:
    """Append ``samples`` as CSV rows.

    On first call, infers the column set from the first sample and
    writes the header. Subsequent calls validate each row's keys
    against that locked set — unknown keys are dropped with a
    one-shot WARN log per unseen key.
    """
    if self._file is None:
        raise RuntimeError("CsvSink: write_many called before open()")
    if not samples:
        return

    rows = [sample_to_row(s) for s in samples]
    if self._writer is None:
        self._columns = tuple(rows[0].keys())
        self._writer = csv.DictWriter(self._file, fieldnames=list(self._columns))
        self._writer.writeheader()

    columns = self._columns
    assert columns is not None  # noqa: S101 — narrow for type checker

    for row in rows:
        unknown = row.keys() - set(columns)
        for key in unknown:
            if key not in self._unknown_columns_warned:
                self._unknown_columns_warned.add(key)
                _logger.warning(
                    "sinks.csv.unknown_column path=%s column=%s action=drop",
                    str(self._path),
                    key,
                )
        filtered = {k: row.get(k) for k in columns}
        self._writer.writerow(filtered)
    self._file.flush()

@classmethod
def failure(cls, error: WatlowError) -> Self:
    """Build a failure result wrapping ``error``."""
    return cls(value=None, error=error)

@classmethod
def success(cls, value: T) -> Self:
    """Build a success result wrapping ``value``."""
    return cls(value=value, error=None)

def merged(self, **updates: Any) -> Self:
    """Return a new context with ``updates`` overlaid. Unknown keys go to ``extra``."""
    known: dict[str, Any] = {}
    extra_updates: dict[str, Any] = {}
    for key, value in updates.items():
        if key in _CONTEXT_KNOWN_FIELDS:
            known[key] = value
        else:
            extra_updates[key] = value

    new_extra: Mapping[str, Any] = (
        MappingProxyType({**self.extra, **extra_updates}) if extra_updates else self.extra
    )
    return replace(self, **known, extra=new_extra)

def add_script(self, command: bytes, reply: ScriptedReply) -> None:
    """Register or overwrite a scripted reply for ``command``."""
    self._script[bytes(command)] = reply

def extend_queue(self, rounds: Iterable[tuple[bytes, ScriptedReply]]) -> None:
    """Append more ordered ``(write, reply)`` pairs to the FIFO queue."""
    self._queue.extend(rounds)

def feed(self, data: bytes) -> None:
    """Push unsolicited bytes into the read buffer.

    Useful for simulating a device that left chatter on the line
    which the protocol client has to drain on recovery.
    """
    self._read_buffer.extend(data)

def force_read_timeout(self, enabled: bool = True) -> None:
    """Force the next read to raise :class:`WatlowTimeoutError`."""
    self._force_read_timeout = enabled

def force_write_timeout(self, enabled: bool = True) -> None:
    """Force the next :meth:`write` to raise :class:`WatlowTimeoutError`."""
    self._force_write_timeout = enabled

@classmethod
def parse(cls, text: str) -> Self:
    """Parse a string like ``"1"``, ``"1.2"``, ``"1.2.3"``, or ``"v1.2"``."""
    m = _PARSE_RE.match(text)
    if m is None:
        raise ValueError(f"invalid firmware version: {text!r}")
    major = int(m.group(1))
    minor = int(m.group(2) or 0)
    patch = int(m.group(3) or 0)
    return cls(major, minor, patch)

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

async def close(self) -> None:
    """Flip the closed flag — no I/O, buffer preserved for inspection."""
    self._closed = True

async def open(self) -> None:
    """No backing resource — just flips the open flag."""
    self._open = True
    self._closed = False

async def write_many(self, samples: Sequence[Sample]) -> None:
    """Append every sample to the internal buffer."""
    if not self.is_open:
        raise RuntimeError("InMemorySink: write_many called before open()")
    self._samples.extend(samples)

async def close(self) -> None:
    """Flush and close the JSONL file. Idempotent."""
    if self._file is None:
        return
    try:
        self._file.flush()
    finally:
        self._file.close()
        self._file = None

async def open(self) -> None:
    """Open the JSONL file for writing in append mode.

    Pre-existing content is preserved; new samples are appended.
    Idempotent on already-open sinks.
    """
    if self._file is not None:
        return
    self._file = self._path.open("a", encoding="utf-8", newline="")

async def write_many(self, samples: Sequence[Sample]) -> None:
    """Serialise each sample as one JSON object per line."""
    if self._file is None:
        raise RuntimeError("JsonlSink: write_many called before open()")
    if not samples:
        return
    for sample in samples:
        row = sample_to_row(sample)
        self._file.write(json.dumps(row, ensure_ascii=False))
        self._file.write("\n")
    self._file.flush()