sartoriuslib.devices¶

async def aclose(self) -> None:
    """Close the underlying transport, if one is wired.

    Idempotent — safe to call multiple times. The factory owns the
    transport's construction and hands it into the session via the
    protocol client; closing the session closes the transport.

    Both clients hold the same transport, so close it once via the
    session's :attr:`transport` accessor rather than once per client
    slot — guards against a future caller that wires both clients
    simultaneously.
    """
    await self.transport.close()

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

def cache_snapshot(self) -> dict[str, int]:
    """Copy of ``cache_key → counter_snapshot`` for test assertions.

    The actual cached values are intentionally not surfaced —
    tests assert on the *presence* and *counter pinning* of an
    entry, not its decoded content.
    """
    return {key: counter for key, (counter, _value) in self._result_cache.items()}

async def cached_execute[Req, Resp](
    self,
    command: Command[Req, Resp],
    request: Req,
    *,
    cache_key: str,
    confirm: bool = False,
    timeout: float | None = None,
) -> Resp:
    """Dispatch ``command`` with a ``0xBA``-keyed result cache.

    Sessions without :attr:`Capability.CONFIG_COUNTER` fall through
    to :meth:`execute`. When the capability is present, the session
    re-reads the config counter before returning a cached value;
    any change flushes the entry and the command re-runs.

    ``cache_key`` is caller-supplied so a command that takes
    arguments (``capacity(area=N)``, ``read_parameter(idx)``) can
    cache separate entries per distinct call.
    """
    if Capability.CONFIG_COUNTER not in self._capabilities:
        return await self.execute(command, request, confirm=confirm, timeout=timeout)
    counter = await self._read_config_counter(timeout=timeout)
    cached = self._result_cache.get(cache_key)
    if cached is not None and cached[0] == counter:
        return cached[1]  # type: ignore[no-any-return]
    result = await self.execute(command, request, confirm=confirm, timeout=timeout)
    self._result_cache[cache_key] = (counter, result)
    return result

def check_state(self) -> None:
    """Raise :class:`SartoriusConnectionError` if the session is BROKEN.

    Public alias for the internal gate run by every dispatch path,
    so lifecycle helpers on :class:`Balance` can reuse the same
    guard without poking at private attributes.
    """
    self._check_state()

async def execute[Req, Resp](
    self,
    command: Command[Req, Resp],
    request: Req,
    *,
    confirm: bool = False,
    timeout: float | None = None,
) -> Resp:
    """Dispatch ``command`` with full pre-I/O gating.

    Gates fire in the design doc §6.1 order; each raise happens
    before any byte is sent, so a gate failure is observably
    equivalent to the call never leaving the host.
    """
    self._check_state()
    self._gate_safety(command, confirm)
    self._gate_protocol(command)
    self._gate_known_denied(command)
    self._gate_priors(command)
    if self._active is ProtocolKind.XBPI:
        return await self._execute_xbpi(command, request, timeout)
    if self._active is ProtocolKind.SBI:
        return await self._execute_sbi(command, request, timeout)
    raise SartoriusError(f"unreachable: session has no active protocol ({self._active!r})")

async def execute_raw_sbi(
    self,
    command: bytes | str,
    *,
    confirm: bool = False,
    timeout: float | None = None,
    expect_lines: int = 1,
) -> SbiReply:
    """Send an arbitrary SBI command token and return the parsed reply."""
    self._check_state()
    if self._active is not ProtocolKind.SBI:
        raise SartoriusProtocolUnsupportedError(
            f"raw_sbi: session is in {self._active.value.upper()} mode",
            context=ErrorContext(protocol=str(self._active.value)),
        )
    if self._sbi is None:
        raise SartoriusError(
            "raw_sbi: session in SBI mode but no SBI client wired",
            context=ErrorContext(protocol="sbi"),
        )
    token = normalize_token(command)
    if token not in SBI_READ_ONLY_TOKENS and not confirm:
        raise SartoriusConfirmationRequiredError(
            f"raw_sbi: token {token!r} is not on the read-only safe-list; "
            "pass confirm=True to proceed",
            context=ErrorContext(
                sbi_token=token,
                protocol="sbi",
                extra={"safe_listed": False},
            ),
        )
    t = timeout if timeout is not None else self._default_timeout
    if self.sbi_autoprint_active and expect_lines > 0:
        self._raise_sbi_autoprint_active("raw_sbi", sbi_token=token)
    reply = await self._sbi.execute(
        token,
        timeout=t,
        command_name="raw_sbi",
        sbi_token=token,
        expect_lines=expect_lines,
    )
    if self._reply_is_surprise_autoprint(reply, sbi_token=token):
        self._sbi.mark_autoprint_active(pending=reply.raw)
        self._raise_sbi_autoprint_active("raw_sbi", sbi_token=token)
    return reply

async def execute_raw_xbpi(
    self,
    opcode: int,
    args: bytes = b"",
    *,
    confirm: bool = False,
    timeout: float | None = None,
) -> XbpiFrame:
    """Send an arbitrary xBPI opcode and return the raw reply frame.

    Bypasses the declarative :class:`Command` layer — the opcode is
    a per-call parameter, so none of the prior / capability gating
    applies. The one hard gate is a safe-list check: opcodes in
    :data:`sartoriuslib.commands.raw.SAFE_READ_ONLY_OPCODES` run
    freely; anything else requires ``confirm=True`` because the
    library cannot know it is safe.

    The availability cache is *not* updated — raw calls are opaque
    at the command-name level.
    """
    self._check_state()
    if self._active is not ProtocolKind.XBPI:
        raise SartoriusProtocolUnsupportedError(
            f"raw_xbpi: session is in {self._active.value.upper()} mode",
            context=ErrorContext(
                opcode=opcode,
                protocol=str(self._active.value),
            ),
        )
    if self._xbpi is None:
        raise SartoriusError(
            "raw_xbpi: session in XBPI mode but no xBPI client wired",
            context=ErrorContext(opcode=opcode, protocol="xbpi"),
        )
    if opcode not in SAFE_READ_ONLY_OPCODES and not confirm:
        raise SartoriusConfirmationRequiredError(
            f"raw_xbpi: opcode 0x{opcode:02x} is not on the read-only safe-list; "
            "pass confirm=True to proceed",
            context=ErrorContext(
                opcode=opcode,
                protocol="xbpi",
                extra={"safe_listed": False},
            ),
        )
    request_bytes = build_command(
        opcode,
        args,
        src_sbn=self._src_sbn,
        dst_sbn=self._dst_sbn,
    )
    t = timeout if timeout is not None else self._default_timeout
    return await self._xbpi.execute(
        request_bytes,
        timeout=t,
        command_name=f"raw_xbpi[0x{opcode:02x}]",
        opcode=opcode,
    )

def invalidate_cache(self, cache_key: str | None = None) -> None:
    """Drop one cached entry, or clear all when ``cache_key`` is ``None``.

    The :class:`Balance` facade calls this after writes whose
    ``0xBA`` bump isn't guaranteed (the §6.3 caveat). Unknown
    keys are a silent no-op — idempotent.
    """
    if cache_key is None:
        self._result_cache.clear()
    else:
        self._result_cache.pop(cache_key, None)

def mark_broken(self) -> None:
    """Transition the session to :attr:`SessionState.BROKEN`.

    Called only from lifecycle operations
    (``Balance.configure_protocol``) when a rollback fails. Once
    broken, every subsequent dispatch refuses with
    :class:`SartoriusConnectionError`.
    """
    self._state = SessionState.BROKEN

async def read_sbi_autoprint_reading(
    self,
    *,
    timeout: float | None = None,
) -> Reading:
    """Read the next valid SBI autoprint weight line without writing."""
    if self._active is not ProtocolKind.SBI:
        raise SartoriusProtocolUnsupportedError(
            f"read_sbi_autoprint_reading: session is in {self._active.value.upper()} mode",
            context=ErrorContext(protocol=str(self._active.value)),
        )
    if self._sbi is None:
        raise SartoriusError(
            "read_sbi_autoprint_reading: session in SBI mode but no SBI client wired",
            context=ErrorContext(protocol="sbi"),
        )
    t = timeout if timeout is not None else self._default_timeout
    deadline = anyio.current_time() + t
    while True:
        remaining = max(0.001, deadline - anyio.current_time())
        reply = await self._sbi.read_line(timeout=remaining)
        try:
            reading = require_reading(reply)
        except SartoriusParseError:
            if anyio.current_time() >= deadline:
                raise
            continue
        self._sbi.mark_autoprint_active()
        return reading

async def read_sbi_line(self, *, timeout: float | None = None) -> SbiReply:
    """Read one unsolicited SBI line, used by autoprint streaming."""
    if self._active is not ProtocolKind.SBI:
        raise SartoriusProtocolUnsupportedError(
            f"read_sbi_line: session is in {self._active.value.upper()} mode",
            context=ErrorContext(protocol=str(self._active.value)),
        )
    if self._sbi is None:
        raise SartoriusError(
            "read_sbi_line: session in SBI mode but no SBI client wired",
            context=ErrorContext(protocol="sbi"),
        )
    t = timeout if timeout is not None else self._default_timeout
    return await self._sbi.read_line(timeout=t)

async def refresh_sbi_autoprint_state(self, *, timeout: float | None = None) -> bool:
    """Passively re-sniff whether SBI autoprint is currently active."""
    if self._active is not ProtocolKind.SBI:
        raise SartoriusProtocolUnsupportedError(
            "refresh_sbi_autoprint_state requires an SBI session",
            context=ErrorContext(protocol=str(self._active.value)),
        )
    if self._sbi is None:
        raise SartoriusError(
            "refresh_sbi_autoprint_state: session in SBI mode but no SBI client wired",
            context=ErrorContext(protocol="sbi"),
        )
    return await self._sbi.refresh_autoprint_state(timeout=timeout)

def replace_clients(
    self,
    *,
    xbpi_client: XbpiProtocolClient | None,
    sbi_client: SbiProtocolClient | None,
    active_protocol: ProtocolKind,
    serial_settings: SerialSettings | None = None,
) -> None:
    """Swap protocol clients atomically — used by ``configure_protocol``.

    The host-side flip closes the old protocol client, reopens the
    transport at new serial framing, builds a new client, and
    verifies. On verification success this method installs the new
    clients and the new active protocol; the availability cache,
    prior warnings, and result cache are all
    cleared because the command surface changes when the protocol
    does (xBPI-only commands have no SBI variant and vice versa,
    and any ``0xBA``-pinned cache entries belong to the old session).

    Refuses to install ``ProtocolKind.AUTO`` — detection must
    resolve to ``XBPI`` or ``SBI`` first. Refuses if the
    corresponding client for ``active_protocol`` is missing.
    """
    if active_protocol is ProtocolKind.AUTO:
        raise SartoriusError(
            "replace_clients: cannot install ProtocolKind.AUTO; "
            "detection must resolve to XBPI or SBI first",
        )
    if active_protocol is ProtocolKind.XBPI and xbpi_client is None:
        raise SartoriusError(
            "replace_clients: active_protocol=XBPI but xbpi_client is None",
        )
    if active_protocol is ProtocolKind.SBI and sbi_client is None:
        raise SartoriusError(
            "replace_clients: active_protocol=SBI but sbi_client is None",
        )
    self._xbpi = xbpi_client
    self._sbi = sbi_client
    self._active = active_protocol
    if serial_settings is not None:
        self._serial_settings = serial_settings
    # Cross-protocol availability and prior warnings do not
    # transfer — clear them so the new protocol starts clean.
    self._availability.clear()
    self._warned_priors.clear()
    self._result_cache.clear()

def set_dst_sbn(self, dst_sbn: int) -> None:
    """Update the destination SBN.

    Used after :meth:`Balance.write_sbn_address` on multidrop links
    where the new address must address the device going forward.
    """
    self._dst_sbn = dst_sbn

def update_identity(
    self,
    *,
    family: BalanceFamily | None = None,
    capabilities: Capability | None = None,
    firmware: FirmwareVersion | None = None,
) -> None:
    """Replace session-level identity state after a live identify call.

    Called by :func:`sartoriuslib.devices.factory.open_device` after
    running the identify commands, so subsequent prior gating sees
    the discovered family and capabilities instead of the
    placeholder ``UNKNOWN`` / empty values.

    Each argument left as ``None`` keeps the existing value.
    """
    if family is not None:
        self._family = family
    if capabilities is not None:
        self._capabilities = capabilities
    if firmware is not None:
        self._firmware = firmware

def __format__(self, format_spec: str) -> str:
    """Delegate format specs to :attr:`value` so ``f"{r:.4f}"`` works.

    The empty spec falls back to :func:`str` (the ``frozen``
    dataclass default) so ``f"{r}"`` still prints the structured
    repr. Off-scale readings (``value`` is ``None``) format as
    ``"None"`` for any non-empty numeric spec rather than raising —
    a stream of mixed valid/None readings is the common case during
    a tare or zero settling window and crashing in a log-line
    f-string would be a surprising failure mode.
    """
    if format_spec == "":
        return str(self)
    if self.value is None:
        return "None"
    return format(self.value, format_spec)

def as_dict(self) -> dict[str, float | int | str | None]:
    """Flatten the reading into a row-shaped dict for tabular sinks.

    Content-only — timing provenance (``received_at``,
    ``monotonic_ns``) lives on the surrounding
    :class:`~sartoriuslib.streaming.sample.Sample` because sample-
    level send/receive boundaries are the authoritative timeline
    (design §10). Booleans render as ``0`` / ``1`` so SQLite picks
    INTEGER affinity and CSV / JSONL round-trip cleanly through
    every stdlib reader.
    """
    return {
        "value": self.value,
        "unit": self.unit.value,
        "sign": self.sign.value,
        "stable": int(self.stable),
        "overload": int(self.overload),
        "underload": int(self.underload),
        "decimals": self.decimals,
        "sequence": self.sequence,
        "protocol": self.protocol.value,
        "raw": self.raw.hex(),
    }

async def aclose(self) -> None:
    """Close the underlying transport. Idempotent."""
    await self._session.aclose()

async def capacity(self, area: int = 0) -> Quantity:
    """Read the weighing capacity for ``area`` (xBPI ``0x0C``).

    Cached by the session when :attr:`Capability.CONFIG_COUNTER`
    is present — the balance's ``0xBA`` counter bumps on display-
    accuracy changes (``p08``) which is the only thing that would
    move this value in practice.

    The wire's typed-float reply does **not** carry a unit byte
    (contrast the 8-byte measurement body's byte [6]). To return
    a complete :class:`Quantity` we read the current display unit
    (``p07``) and fold it in here. ``get_display_unit()`` is itself
    cached on ``0xBA`` via the parameter-table cache, so two
    successive ``capacity()`` calls only re-read ``0xBA`` (twice,
    once per call) and not ``0x0C`` or ``0x55``. If
    ``get_display_unit()`` itself fails (e.g. the parameter table
    is unreachable), the unit falls back to :attr:`Unit.UNKNOWN`
    and the numeric value still returns — fail-open is more useful
    than a hard error for a metadata read.
    """
    raw = await self._session.cached_execute(
        READ_CAPACITY,
        MetrologyRequest(area=area),
        cache_key=f"capacity:{area}",
    )
    return await self._resolve_metrology_unit(raw)

async def configure_protocol(
    self,
    target: ProtocolKind,
    *,
    baudrate: int | None = None,
    parity: Parity | None = None,
    stopbits: StopBits | None = None,
    timeout: float | None = None,
    confirm: bool = False,
) -> DeviceInfo:
    """Switch this balance's active wire protocol (and optionally framing).

    ``DANGEROUS`` — requires ``confirm=True``. The flip is purely
    host-side: per ``docs/protocol.md`` §2.1 the device's protocol
    mode changes via the front-panel menu, never via xBPI for the
    PC-USB port (verified empirically 2026-04-25 on MSE1203S — the
    PC-USB protocol selector is not in the xBPI parameter table at
    all; only `Device → PC-USB → Dat.Rec.` on the front panel
    flips it). On the 9-pin peripheral port, `p35` write +
    SAVE_MENU + cold boot does work programmatically, but most
    users are on PC-USB.

    This method reconciles the host with the user's front-panel
    change by closing the current protocol client, reopening the
    transport at the new serial framing (any ``None`` argument keeps
    the existing value), and building the new client. It then
    verifies via an identity probe and refreshes :attr:`info` from
    the new protocol.

    If the user has NOT actually flipped the front panel before
    calling this method, the post-switch identity probe will fail
    — typically with a :class:`SartoriusFrameError` ("bad marker
    byte 0x20") when xBPI is requested but the wire is still
    emitting SBI autoprint, or a timeout when SBI is requested
    but the wire is still on xBPI. Treat that error as a strong
    signal that the front-panel mode does not match ``target``.

    On any failure during the switch the method attempts to roll
    back to the original serial framing. If that rollback also
    fails the underlying :class:`Session` transitions to
    :attr:`SessionState.BROKEN` and a
    :class:`SartoriusConnectionError` is raised — the caller must
    close this balance and re-open via
    :func:`sartoriuslib.open_device` to recover.

    Same-protocol no-op: when ``target`` equals the current
    protocol and no framing override is supplied, the call returns
    the cached :class:`DeviceInfo` (or runs :meth:`identify` if
    none has been cached yet) without touching the transport.
    """
    if target is ProtocolKind.AUTO:
        raise SartoriusValidationError(
            "configure_protocol: target must be XBPI or SBI, not AUTO",
            context=ErrorContext(
                command_name="configure_protocol",
                extra={"target": "auto"},
            ),
        )
    if not confirm:
        raise SartoriusConfirmationRequiredError(
            "configure_protocol is DANGEROUS; pass confirm=True to execute",
            context=ErrorContext(
                command_name="configure_protocol",
                extra={"target": target.value},
            ),
        )

    session = self._session
    session.check_state()

    no_framing_change = baudrate is None and parity is None and stopbits is None
    if target is session.active_protocol and no_framing_change:
        return self._info if self._info is not None else await self.identify()

    transport = session.transport

    old_settings = session.serial_settings
    old_xbpi = session.xbpi_client
    old_sbi = session.sbi_client
    old_active = session.active_protocol
    active_lock = (
        old_xbpi.lock
        if old_active is ProtocolKind.XBPI and old_xbpi is not None
        else old_sbi.lock
        if old_sbi is not None
        else None
    )
    if active_lock is None:
        raise SartoriusError(
            "configure_protocol: active protocol has no client lock",
            context=ErrorContext(command_name="configure_protocol"),
        )

    t = timeout if timeout is not None else session.default_timeout

    async with active_lock:
        try:
            await transport.drain_input()
            await transport.reopen(
                baudrate=baudrate,
                parity=parity,
                stopbits=stopbits,
            )
            new_xbpi: XbpiProtocolClient | None = None
            new_sbi: SbiProtocolClient | None = None
            if target is ProtocolKind.XBPI:
                new_xbpi = XbpiProtocolClient(transport, default_timeout=t)
            else:
                new_sbi = SbiProtocolClient(transport, default_timeout=t)
                await new_sbi.detect_autoprint(timeout=min(t, 0.25))
            if not (
                target is ProtocolKind.SBI and new_sbi is not None and new_sbi.autoprint_active
            ):
                await _verify_identity_probe(
                    target,
                    new_xbpi,
                    new_sbi,
                    src_sbn=session.src_sbn,
                    dst_sbn=session.dst_sbn,
                    timeout=t,
                )
            new_settings = _overlay_settings(
                old_settings,
                baudrate=baudrate,
                parity=parity,
                stopbits=stopbits,
            )
            session.replace_clients(
                xbpi_client=new_xbpi,
                sbi_client=new_sbi,
                active_protocol=target,
                serial_settings=new_settings,
            )
            _dispose_replaced_clients(
                old_xbpi=old_xbpi,
                old_sbi=old_sbi,
                new_xbpi=new_xbpi,
                new_sbi=new_sbi,
            )
        except Exception as switch_error:
            await self._rollback_configure_protocol(
                transport=transport,
                old_settings=old_settings,
                target=target,
                cause=switch_error,
            )
            raise

    # Identify outside the old client's lock — Session.execute will
    # acquire the new client's lock itself.
    return await self.identify()

async def discover_temperature_sensors(
    self,
    *,
    max_index: int = _TEMPERATURE_DISCOVERY_MAX_INDEX,
) -> tuple[int, ...]:
    """Probe the device for installed temperature sensors at runtime.

    Iterates indices ``0..max_index`` calling :meth:`temperature`
    on each. Records every index that replies — both real
    readings (``celsius`` is a ``float``) **and** sentinel slots
    (``celsius`` is ``None`` because the firmware returned
    ``7f ff ff ff``). Stops early on
    :class:`SartoriusIndexOutOfRangeError`, which the device
    emits past the last valid index. Updates the cached
    :class:`DeviceInfo`'s
    :attr:`temperature_sensor_indices` with the discovered tuple
    and returns it.

    Device-agnostic by design — no family table is consulted, so
    a balance we have never tested still produces an honest
    sensor map.

    ``max_index`` is a safety cap, not a contract: 8 is the
    default headroom (Cubis MSE captures stop at 3). Probing a
    device with no sensors produces an empty tuple. Each probe
    is one round-trip; the result is not cached on ``0xBA``
    because per-call temperature reads are not cached either —
    callers re-discover on demand.
    """
    if max_index < 0:
        raise ValueError(f"max_index must be >= 0, got {max_index}")
    # Local import — lazy to avoid pulling errors module at
    # construction time. The class is intentionally module-level.
    from sartoriuslib.errors import (  # noqa: PLC0415
        SartoriusIndexOutOfRangeError,
        SartoriusUnsupportedCommandError,
    )

    discovered: list[int] = []
    for sensor in range(max_index + 1):
        try:
            reading = await self.temperature(sensor)
        except SartoriusIndexOutOfRangeError:
            # Past the last valid index — clean stop.
            break
        except SartoriusUnsupportedCommandError:
            # Some firmwares mis-report end-of-list as 0x04 instead
            # of 0x10. ``temperature`` is parameterized so the
            # availability cache is not poisoned for in-range
            # indices; we still treat it as the end signal here.
            break
        discovered.append(reading.sensor)
    result = tuple(discovered)
    if self._info is not None:
        self._info = dataclasses.replace(
            self._info,
            temperature_sensor_indices=result,
        )
    return result

async def get_auto_zero(self) -> AutoZeroMode:
    """Read ``p06`` (auto-zero tracking) as an :class:`AutoZeroMode`."""
    spec = PARAMETER_TABLE[_TYPED_ACCESSOR_INDICES["auto_zero"]]
    current = await self._get_typed(spec.index)
    return cast("AutoZeroMode", spec.decode(current))

async def get_display_unit(self) -> Unit:
    """Read ``p07`` (display unit) as a :class:`Unit`."""
    spec = PARAMETER_TABLE[_TYPED_ACCESSOR_INDICES["display_unit"]]
    current = await self._get_typed(spec.index)
    # p07 is the only unit-valued spec; decode() returns a Unit.
    return cast("Unit", spec.decode(current))

async def get_filter_mode(self) -> FilterMode:
    """Read ``p01`` (filter mode) as a :class:`FilterMode`."""
    spec = PARAMETER_TABLE[_TYPED_ACCESSOR_INDICES["filter_mode"]]
    current = await self._get_typed(spec.index)
    # The spec's enum class is FilterMode — decode() is guaranteed
    # to return a FilterMode member (or FilterMode.UNKNOWN). The
    # signature is the union across all specs so we narrow here.
    return cast("FilterMode", spec.decode(current))

async def get_isocal_mode(self) -> IsoCalMode:
    """Read ``p15`` (isoCAL mode) as an :class:`IsoCalMode` (Cubis only)."""
    spec = PARAMETER_TABLE[_TYPED_ACCESSOR_INDICES["isocal_mode"]]
    current = await self._get_typed(spec.index)
    return cast("IsoCalMode", spec.decode(current))

async def get_menu_access(self) -> MenuAccessMode:
    """Read ``p40`` (front-panel menu lock) as :class:`MenuAccessMode`."""
    spec = PARAMETER_TABLE[_TYPED_ACCESSOR_INDICES["menu_access"]]
    current = await self._get_typed(spec.index)
    return cast("MenuAccessMode", spec.decode(current))