sartoriuslib.protocol¶

async def detect_protocol(
    transport: Transport,
    *,
    timeout: float = _DEFAULT_PROBE_TIMEOUT,
    sniff_window: float = _DEFAULT_SNIFF_WINDOW,
    src_sbn: int = HOST_SBN_DEFAULT,
    dst_sbn: int = BALANCE_SBN_DEFAULT,
) -> DetectionResult:
    """Detect xBPI vs SBI on an already-open ``transport``.

    Runs the conservative sequence from design §4.3 in order — drain →
    passive sniff → xBPI probe → SBI probe → fail. Each probe writes at
    most one frame. The transport's serial settings are never changed.

    Arguments:
        transport: An open :class:`Transport`. Caller owns lifecycle.
        timeout: Per-probe timeout for the xBPI and SBI identity probes.
        sniff_window: Passive listen window for SBI autoprint, in seconds.
        src_sbn: Source SBN for the xBPI probe frame (``0x01`` host
            convention by default).
        dst_sbn: Destination SBN for the xBPI probe frame (``0x09``
            balance factory default by default).

    Returns:
        A :class:`DetectionResult` whose ``protocol`` is ``XBPI`` or
        ``SBI``. When ``autoprint_active`` is ``True``, ``pending_lines``
        carries the sniffed bytes for re-queue.

    Raises:
        SartoriusError: No xBPI or SBI device responded — neither the
            passive sniff nor either probe produced a recognisable reply.
            Hard transport faults (e.g. the port closed mid-detect)
            propagate as :class:`SartoriusConnectionError` unchanged.
    """
    label = transport.label

    # 1. Passive autoprint sniff. We deliberately do NOT drain first: a
    #    balance left in autoprint mode may have a complete line already
    #    sitting in the OS buffer when we connect, and the design promises
    #    we will not drop the first sample. Stale partial bytes (no CRLF)
    #    can't fool the sniff — read_until times out without consuming them
    #    and the pre-probe drain below clears them before xBPI runs.
    autoprint, sniffed = await _sniff_for_autoprint(transport, sniff_window)
    if autoprint:
        return DetectionResult(
            protocol=ProtocolKind.SBI,
            autoprint_active=True,
            pending_lines=sniffed,
        )

    # 2. Drain anything left over from the sniff (partial lines, stray
    #    bytes that didn't form CRLF) so the xBPI length-prefix read starts
    #    from a clean buffer.
    with contextlib.suppress(SartoriusError):
        await transport.drain_input()

    # 3. xBPI probe — READ_MODEL. A valid frame (even one carrying an error
    #    subtype) confirms xBPI: only an xBPI device builds a length-prefixed
    #    marker-tagged frame.
    if await _probe_xbpi(transport, timeout=timeout, src_sbn=src_sbn, dst_sbn=dst_sbn):
        return DetectionResult(protocol=ProtocolKind.XBPI)

    with contextlib.suppress(SartoriusError):
        await transport.drain_input()

    # 4. SBI identity probe. Any CRLF-terminated reply counts as evidence —
    #    content interpretation is the SBI parser's job, not detection's.
    if await _probe_sbi(transport, timeout=timeout):
        return DetectionResult(protocol=ProtocolKind.SBI)

    # 5. Clear failure. No opcode sweeps, no fallback baud rates.
    raise SartoriusError(
        f"auto-detect: no responsive xBPI or SBI device on {label!r} "
        f"(sniff {sniff_window}s, probe timeout {timeout}s)",
        context=ErrorContext(
            port=label,
            command_name="auto_detect",
            extra={"sniff_window_s": sniff_window, "probe_timeout_s": timeout},
        ),
    )

def make_protocol_client(
    protocol: ProtocolKind,
    transport: Transport,
    *,
    default_timeout: float = 1.0,
) -> XbpiProtocolClient | SbiProtocolClient:
    """Build the concrete protocol client for ``protocol`` on ``transport``.

    :attr:`AUTO` is never valid at factory time — the detection step
    resolves it first.
    """
    if protocol is ProtocolKind.XBPI:
        return XbpiProtocolClient(transport, default_timeout=default_timeout)
    if protocol is ProtocolKind.SBI:
        return SbiProtocolClient(transport, default_timeout=default_timeout)
    raise ValueError(
        f"cannot build a protocol client for {protocol!r}; "
        "AUTO must resolve to XBPI or SBI before this point",
    )

def make_protocol_client(
    protocol: ProtocolKind,
    transport: Transport,
    *,
    default_timeout: float = 1.0,
) -> XbpiProtocolClient | SbiProtocolClient:
    """Build the concrete protocol client for ``protocol`` on ``transport``.

    :attr:`AUTO` is never valid at factory time — the detection step
    resolves it first.
    """
    if protocol is ProtocolKind.XBPI:
        return XbpiProtocolClient(transport, default_timeout=default_timeout)
    if protocol is ProtocolKind.SBI:
        return SbiProtocolClient(transport, default_timeout=default_timeout)
    raise ValueError(
        f"cannot build a protocol client for {protocol!r}; "
        "AUTO must resolve to XBPI or SBI before this point",
    )

async def detect_protocol(
    transport: Transport,
    *,
    timeout: float = _DEFAULT_PROBE_TIMEOUT,
    sniff_window: float = _DEFAULT_SNIFF_WINDOW,
    src_sbn: int = HOST_SBN_DEFAULT,
    dst_sbn: int = BALANCE_SBN_DEFAULT,
) -> DetectionResult:
    """Detect xBPI vs SBI on an already-open ``transport``.

    Runs the conservative sequence from design §4.3 in order — drain →
    passive sniff → xBPI probe → SBI probe → fail. Each probe writes at
    most one frame. The transport's serial settings are never changed.

    Arguments:
        transport: An open :class:`Transport`. Caller owns lifecycle.
        timeout: Per-probe timeout for the xBPI and SBI identity probes.
        sniff_window: Passive listen window for SBI autoprint, in seconds.
        src_sbn: Source SBN for the xBPI probe frame (``0x01`` host
            convention by default).
        dst_sbn: Destination SBN for the xBPI probe frame (``0x09``
            balance factory default by default).

    Returns:
        A :class:`DetectionResult` whose ``protocol`` is ``XBPI`` or
        ``SBI``. When ``autoprint_active`` is ``True``, ``pending_lines``
        carries the sniffed bytes for re-queue.

    Raises:
        SartoriusError: No xBPI or SBI device responded — neither the
            passive sniff nor either probe produced a recognisable reply.
            Hard transport faults (e.g. the port closed mid-detect)
            propagate as :class:`SartoriusConnectionError` unchanged.
    """
    label = transport.label

    # 1. Passive autoprint sniff. We deliberately do NOT drain first: a
    #    balance left in autoprint mode may have a complete line already
    #    sitting in the OS buffer when we connect, and the design promises
    #    we will not drop the first sample. Stale partial bytes (no CRLF)
    #    can't fool the sniff — read_until times out without consuming them
    #    and the pre-probe drain below clears them before xBPI runs.
    autoprint, sniffed = await _sniff_for_autoprint(transport, sniff_window)
    if autoprint:
        return DetectionResult(
            protocol=ProtocolKind.SBI,
            autoprint_active=True,
            pending_lines=sniffed,
        )

    # 2. Drain anything left over from the sniff (partial lines, stray
    #    bytes that didn't form CRLF) so the xBPI length-prefix read starts
    #    from a clean buffer.
    with contextlib.suppress(SartoriusError):
        await transport.drain_input()

    # 3. xBPI probe — READ_MODEL. A valid frame (even one carrying an error
    #    subtype) confirms xBPI: only an xBPI device builds a length-prefixed
    #    marker-tagged frame.
    if await _probe_xbpi(transport, timeout=timeout, src_sbn=src_sbn, dst_sbn=dst_sbn):
        return DetectionResult(protocol=ProtocolKind.XBPI)

    with contextlib.suppress(SartoriusError):
        await transport.drain_input()

    # 4. SBI identity probe. Any CRLF-terminated reply counts as evidence —
    #    content interpretation is the SBI parser's job, not detection's.
    if await _probe_sbi(transport, timeout=timeout):
        return DetectionResult(protocol=ProtocolKind.SBI)

    # 5. Clear failure. No opcode sweeps, no fallback baud rates.
    raise SartoriusError(
        f"auto-detect: no responsive xBPI or SBI device on {label!r} "
        f"(sniff {sniff_window}s, probe timeout {timeout}s)",
        context=ErrorContext(
            port=label,
            command_name="auto_detect",
            extra={"sniff_window_s": sniff_window, "probe_timeout_s": timeout},
        ),
    )

def build_command(
    opcode: int,
    args: bytes = b"",
    *,
    src_sbn: int = HOST_SBN_DEFAULT,
    dst_sbn: int = BALANCE_SBN_DEFAULT,
) -> bytes:
    """Assemble a host→balance frame.

    Arguments:
        opcode: Command byte (``0x00`` – ``0xFF``).
        args: Pre-encoded argument bytes (usually one or more TLVs — see
            :mod:`sartoriuslib.protocol.xbpi.tlv`). Empty for no-arg
            commands.
        src_sbn: Source SBN; defaults to the host convention ``0x01``.
        dst_sbn: Destination SBN; defaults to the balance factory
            default ``0x09``.

    Returns:
        The complete frame bytes, length-prefixed and checksummed, ready
        to hand to :meth:`Transport.write`.
    """
    _require_byte(opcode, "opcode")
    _require_byte(src_sbn, "src_sbn")
    _require_byte(dst_sbn, "dst_sbn")
    payload = bytes([src_sbn, dst_sbn, opcode]) + bytes(args)
    # ``length`` counts every byte that will follow the length byte —
    # including the not-yet-appended checksum.
    length = len(payload) + 1
    if length > _MAX_LENGTH:
        raise SartoriusFrameError(
            f"frame too long: {length + 1} bytes (max 256)",
            context=ErrorContext(opcode=opcode, extra={"length": length}),
        )
    pre_chk = bytes([length]) + payload
    return pre_chk + bytes([checksum(pre_chk)])

def checksum(data: bytes) -> int:
    """Return ``sum(data) & 0xFF`` — the xBPI frame checksum.

    Trivial by design. No CRC, no seed; see ``docs/protocol.md`` §12.
    """
    return sum(data) & 0xFF

def parse_frame(data: bytes) -> XbpiFrame:
    """Validate and decompose a balance→host frame.

    Raises:
        SartoriusFrameError: Frame too short, length byte inconsistent
            with buffer size, marker is not ``0x41``, or checksum
            mismatch.
    """
    raw = bytes(data)
    if len(raw) < MIN_FRAME_SIZE:
        raise SartoriusFrameError(
            f"frame too short: got {len(raw)} bytes (min {MIN_FRAME_SIZE})",
            context=ErrorContext(raw_response=raw),
        )
    length = raw[0]
    expected_total = length + 1
    if len(raw) != expected_total:
        raise SartoriusFrameError(
            f"frame length mismatch: length byte says {expected_total} bytes, got {len(raw)}",
            context=ErrorContext(
                raw_response=raw,
                extra={"declared_length": length, "buffer_size": len(raw)},
            ),
        )
    marker = raw[1]
    if marker != RX_MARKER:
        raise SartoriusFrameError(
            f"bad marker byte 0x{marker:02x} (expected 0x{RX_MARKER:02x})",
            context=ErrorContext(raw_response=raw, extra={"marker": marker}),
        )
    subtype = raw[2]
    body = raw[3:-1]
    chk = raw[-1]
    expected_chk = checksum(raw[:-1])
    if chk != expected_chk:
        raise SartoriusFrameError(
            f"bad checksum 0x{chk:02x} (expected 0x{expected_chk:02x})",
            context=ErrorContext(
                raw_response=raw,
                extra={"checksum": chk, "expected_checksum": expected_chk},
            ),
        )
    return XbpiFrame(
        length=length,
        marker=marker,
        subtype=subtype,
        body=bytes(body),
        checksum=chk,
        raw=raw,
    )

def body_length_for_subtype(subtype: int) -> int | None:
    """Expected body length for ``subtype``, or ``None`` if variable.

    Implements the §4 formula: "high nibble = type class, low nibble =
    body length for types 0..4; for type 5, length = 16 + low". Returns
    ``None`` for subtypes whose body length is genuinely variable
    (notably ``0x48`` which carries either 8 or 17 bytes, and ``0x00``
    which usually carries 0 but occasionally a variable body — e.g.
    opcode ``0xBC`` module list).
    """
    # 0x48 measurement: 8 *or* 17 — caller disambiguates by the length
    # byte on the outer frame. 0x00 ACK: usually empty but 0xBC's module
    # list rides this subtype with a variable body. Both are variable.
    if subtype in _VARIABLE_BODY_SUBTYPES:
        return None
    high = subtype & 0xF0
    low = subtype & 0x0F
    if high in _FIXED_LOW_NIBBLE_LENGTH_FAMILIES:
        return low
    if high == _LONG_STRING_FAMILY_HIGH:
        return _LONG_STRING_FAMILY_OFFSET + low
    return None

def subtype_family(subtype: int) -> SubtypeFamily:
    """Return the family classifier for a reply subtype byte.

    Falls back to high-nibble dispatch for subtypes not explicitly
    tabulated, and to :attr:`SubtypeFamily.UNKNOWN` when even that fails.
    """
    if subtype in _KNOWN_SUBTYPES:
        return _KNOWN_SUBTYPES[subtype]
    return _HIGH_NIBBLE_FAMILIES.get(subtype & 0xF0, SubtypeFamily.UNKNOWN)

def decode_tlv(data: bytes, offset: int = 0) -> tuple[int, bytes, int]:
    """Decode one TLV record starting at ``offset``.

    Returns ``(tag, value_bytes, next_offset)``. Raises
    :class:`SartoriusFrameError` on unknown tags or truncated values.
    """
    if offset >= len(data):
        raise SartoriusFrameError(
            "TLV sequence truncated at tag position",
            context=ErrorContext(extra={"offset": offset, "total": len(data)}),
        )
    tag = data[offset]
    if tag not in TLV_TAG_SIZES:
        raise SartoriusFrameError(
            f"unknown TLV tag 0x{tag:02x} at offset {offset}",
            context=ErrorContext(extra={"tag": tag, "offset": offset}),
        )
    size = TLV_TAG_SIZES[tag]
    value_start = offset + 1
    value_end = value_start + size
    if value_end > len(data):
        raise SartoriusFrameError(
            f"TLV value for tag 0x{tag:02x} truncated "
            f"(needed {size} bytes, have {len(data) - value_start})",
            context=ErrorContext(extra={"tag": tag, "offset": offset}),
        )
    return tag, bytes(data[value_start:value_end]), value_end

def encode_tlv(tag: int, value: int | bytes) -> bytes:
    """Encode a single TLV record.

    ``value`` may be an ``int`` (encoded big-endian into the tag's size)
    or raw ``bytes`` (emitted verbatim, length-checked against the tag).
    """
    if tag not in TLV_TAG_SIZES:
        raise SartoriusFrameError(
            f"unknown TLV tag 0x{tag:02x}",
            context=ErrorContext(extra={"tag": tag}),
        )
    size = TLV_TAG_SIZES[tag]
    if isinstance(value, int):
        if value < 0:
            raise SartoriusFrameError(
                f"TLV value must be non-negative (got {value})",
                context=ErrorContext(extra={"tag": tag, "value": value}),
            )
        try:
            payload = value.to_bytes(size, "big")
        except OverflowError as exc:
            raise SartoriusFrameError(
                f"TLV value {value} does not fit in {size} byte(s) for tag 0x{tag:02x}",
                context=ErrorContext(extra={"tag": tag, "value": value}),
            ) from exc
        return bytes([tag]) + payload
    if len(value) != size:
        raise SartoriusFrameError(
            f"TLV value for tag 0x{tag:02x} must be {size} byte(s), got {len(value)}",
            context=ErrorContext(extra={"tag": tag, "value_len": len(value)}),
        )
    return bytes([tag]) + bytes(value)

def parse_tlv_sequence(body: bytes) -> list[tuple[int, bytes]]:
    """Walk ``body`` as a concatenation of TLV records.

    Returns a list of ``(tag, value_bytes)`` tuples. Raises
    :class:`SartoriusFrameError` if any record is truncated or any tag is
    unknown.

    Note: parameter-table replies (opcode ``0x55``) have the subtype byte
    double as the first TLV tag per §5.3 — the caller must prepend the
    subtype byte before passing the body here.
    """
    out: list[tuple[int, bytes]] = []
    offset = 0
    while offset < len(body):
        tag, value, offset = decode_tlv(body, offset)
        out.append((tag, value))
    return out

def tlv_value_as_int(value: bytes) -> int:
    """Decode a TLV value as a big-endian unsigned integer."""
    return int.from_bytes(value, "big")

def decode_decimals(byte5: int) -> int:
    """Return the displayed decimal-place count encoded in byte [5].

    High nibble = decimals (``docs/protocol.md`` §8.4). The low nibble
    has a WZA-mg quirk that is *advisory* and ignored here.
    """
    return (byte5 >> 4) & 0x0F

def decode_sign(byte6: int) -> Sign:
    """Decode the sign bits (top 2) of byte [6].

    ``0x00`` = exactly zero, ``0x40`` = positive, ``0x80`` = negative.
    Any other combination (shouldn't occur — only one bit of the 2-bit
    field is set at a time on the wire) decodes to :attr:`Sign.UNKNOWN`.
    """
    bits = byte6 & SIGN_MASK
    if bits == SIGN_BITS_ZERO:
        return Sign.ZERO
    if bits == SIGN_BITS_POSITIVE:
        return Sign.POSITIVE
    if bits == SIGN_BITS_NEGATIVE:
        return Sign.NEGATIVE
    return Sign.UNKNOWN

def decode_unit(byte6: int) -> Unit:
    """Full decode of byte [6] to a :class:`Unit` (strips the sign bits)."""
    return unit_byte_to_unit(byte6 & UNIT_ID_MASK)

def unit_byte_to_unit(unit_id: int) -> Unit:
    """Map a 6-bit base-unit ID to a :class:`Unit`.

    Unknown IDs decode to :attr:`Unit.UNKNOWN` — never raises — because
    new unit codes are expected as more of the parameter-table display
    enum gets mapped (see ``docs/protocol.md`` §10.1 index 7).
    """
    return _UNIT_ID_TO_UNIT.get(unit_id & UNIT_ID_MASK, Unit.UNKNOWN)

def decode_error_body(body: bytes) -> ErrorBody:
    """Decode a 1-byte error body into :class:`ErrorBody`.

    The single byte is the device error code (``0x03`` / ``0x04`` /
    ``0x06`` / ``0x07`` / ``0x10`` / ``0x11`` — see
    ``docs/protocol.md`` §6). Mapping to typed exceptions happens at
    the session layer; this decoder stays neutral.
    """
    if len(body) != _ERROR_BODY_LEN:
        raise SartoriusParseError(
            f"error body must be {_ERROR_BODY_LEN} byte, got {len(body)}",
            context=ErrorContext(raw_response=bytes(body)),
        )
    return ErrorBody(code=body[0])

def decode_long_measurement_body(body: bytes) -> LongMeasurementBody:
    """Decode a 17-byte long streaming measurement (§8.3).

    Layout is short measurement (8 B) + delimiter (1 B, always ``0x48``)
    + status block (8 B). The delimiter matches the subtype byte by
    coincidence.
    """
    if len(body) != _LONG_MEASUREMENT_LEN:
        raise SartoriusParseError(
            f"long measurement body must be {_LONG_MEASUREMENT_LEN} bytes, got {len(body)}",
            context=ErrorContext(raw_response=bytes(body)),
        )
    raw = bytes(body)
    measurement = decode_measurement_body(raw[0:_SHORT_MEASUREMENT_LEN])
    delimiter = raw[_SHORT_MEASUREMENT_LEN]
    if delimiter != _LONG_MEASUREMENT_DELIMITER:
        raise SartoriusParseError(
            f"long measurement delimiter must be 0x{_LONG_MEASUREMENT_DELIMITER:02x}, "
            f"got 0x{delimiter:02x}",
            context=ErrorContext(raw_response=raw),
        )
    status = decode_status_block_body(raw[_SHORT_MEASUREMENT_LEN + 1 :])
    return LongMeasurementBody(
        measurement=measurement,
        delimiter=delimiter,
        status=status,
    )

def decode_measurement_body(body: bytes) -> MeasurementBody:
    """Decode an 8-byte short measurement body.

    The value is ``float32`` big-endian in bytes [0..3]. Bytes [5..6]
    pack decimals (high nibble of [5]), sign (top 2 bits of [6]), and
    the base-unit ID (low 6 bits of [6]). Byte [7]'s ``0x40`` bit marks
    a stable reading universally across families. An off-scale reading
    presents the ``7f ff ff ff ff`` sentinel in bytes [0..4]; the
    status block disambiguates overload vs underload, so this decoder
    just reports ``off_scale`` and leaves ``value`` as ``None``.

    .. note::

        The MSE Cubis emits the same ``7f ff ff ff ff`` sentinel for
        ~6 frames (~2 s) immediately after :meth:`Balance.zero` while
        the cell recomputes its zero point. The wire is ambiguous —
        from the body alone we can't tell "cell busy" apart from
        "overload" or "underload". Callers that need that distinction
        invoke :meth:`Balance.status` (xBPI ``0x32``); the status
        block's ``state`` byte carries the disambiguation
        (``0x82``/``0x84`` for overload/underload, plain stable/
        unstable for a settling cell). Observed on hardware day —
        :class:`Reading` faithfully reflects ``value=None`` /
        ``off_scale=True`` either way.
    """
    if len(body) != _SHORT_MEASUREMENT_LEN:
        raise SartoriusParseError(
            f"short measurement body must be {_SHORT_MEASUREMENT_LEN} bytes, got {len(body)}",
            context=ErrorContext(raw_response=bytes(body)),
        )
    raw = bytes(body)
    off_scale = raw[0:5] == OFF_SCALE_SENTINEL
    if off_scale:
        value: float | None = None
    else:
        value = struct.unpack(">f", raw[0:4])[0]
    aux = raw[4]
    byte5 = raw[5]
    byte6 = raw[6]
    flags = raw[7]
    decimals = decode_decimals(byte5)
    sign = decode_sign(byte6)
    unit = decode_unit(byte6)
    stable = bool(flags & STABLE_FLAG)
    return MeasurementBody(
        raw=raw,
        value=value,
        aux=aux,
        decimals=decimals,
        unit=unit,
        sign=sign,
        stable=stable,
        off_scale=off_scale,
        unit_raw=byte6,
        flags=flags,
    )

def decode_status_block_body(body: bytes) -> StatusBlockBody:
    """Decode an 8-byte status block.

    See ``docs/protocol.md`` §8.2. The portable stability bit is
    ``state & 0x08`` — Cubis encodes ``state=0x88`` stable vs ``0x80``
    unstable (the ``0x80`` base marks "measurement valid"), while
    WZA/BCE use ``state=0x08`` vs ``0x00``. Overload (state ``0x82``)
    and underload (``0x84``) have only been captured on Cubis; on
    non-Cubis this decoder still reports them when the exact pattern
    matches, and ``False`` otherwise.

    ``adc_trusted`` and ``isocal_due`` are MSE-only; decoded as ``None``
    when the frame does not look like a Cubis-style status block (base
    bit ``0x80`` clear in ``state``).
    """
    if len(body) != _STATUS_BLOCK_LEN:
        raise SartoriusParseError(
            f"status block must be {_STATUS_BLOCK_LEN} bytes, got {len(body)}",
            context=ErrorContext(raw_response=bytes(body)),
        )
    raw = bytes(body)
    aux_flag = raw[0]
    state = raw[3]
    status = raw[4]
    sequence = raw[7]
    stable = bool(state & _STATE_STABLE_BIT)
    overload = state == _STATE_OVERLOAD
    underload = state == _STATE_UNDERLOAD
    # Cubis signature: the "measurement valid" base bit 0x80 is set on
    # stable/unstable/overload/underload states (0x88/0x80/0x82/0x84).
    # Non-Cubis never sets 0x80, so treat ``adc_trusted``/``isocal_due``
    # as Cubis-only signals.
    cubis_shape = bool(state & _CUBIS_STATE_BASE)
    adc_trusted: bool | None = bool(status & _STATUS_ADC_TRUSTED) if cubis_shape else None
    isocal_due: bool | None = bool(status & _STATUS_ISOCAL_DUE) if cubis_shape else None
    return StatusBlockBody(
        raw=raw,
        aux_flag=aux_flag,
        state=state,
        status=status,
        sequence=sequence,
        stable=stable,
        overload=overload,
        underload=underload,
        adc_trusted=adc_trusted,
        isocal_due=isocal_due,
    )

def decode_typed_float_body(body: bytes) -> TypedFloatBody:
    """Decode a 5-byte typed-float body: ``float32 BE`` + 1-byte aux."""
    if len(body) != _TYPED_FLOAT_LEN:
        raise SartoriusParseError(
            f"typed_float body must be {_TYPED_FLOAT_LEN} bytes, got {len(body)}",
            context=ErrorContext(raw_response=bytes(body)),
        )
    raw = bytes(body)
    value = struct.unpack(">f", raw[0:4])[0]
    aux = raw[4]
    return TypedFloatBody(value=value, aux=aux)

def is_status_block_body(body: bytes) -> bool:
    """Heuristic: does this 8-byte body look like a status block?

    Short measurements and status blocks share subtype ``0x48`` and both
    carry 8-byte bodies. Disambiguation uses the §8.2 marker pattern:
    bytes [1..2] == ``00 81`` and bytes [5..6] == ``10 00`` reliably
    identify a status block; short measurements only match by accident.
    """
    return (
        len(body) == _STATUS_BLOCK_LEN
        and body[1:3] == STATUS_BLOCK_MARKER_PREFIX
        and body[5:7] == STATUS_BLOCK_MARKER_SUFFIX
    )

def build_command(command: bytes | str, *, terminator: bytes = b"") -> bytes:
    """Build an SBI command token, optionally appending a terminator."""
    token = normalize_token(command)
    return token + terminator

def is_autoprint_line(line: bytes | str) -> bool:
    """Return ``True`` when ``line`` looks like an unsolicited weight line."""
    raw = line.encode("ascii", errors="ignore") if isinstance(line, str) else line
    body = strip_line_terminator(raw).decode("ascii", errors="replace")
    text = body.strip()
    if not text:
        return False
    if _BARE_NUMBER_RE.fullmatch(body) and (
        body == text or len(body) < _MIN_FORMATTED_WEIGHT_BODY_CHARS
    ):
        return False
    marker = " ".join(text.upper().split())
    if marker.startswith("STAT "):
        marker = marker.removeprefix("STAT ").strip()
    return _AUTOPRINT_RE.match(body) is not None or marker in _AUTOPRINT_SPECIALS

def normalize_token(command: bytes | str) -> bytes:
    """Normalize a user-facing SBI command into on-wire bytes.

    Accepts raw bytes, strings containing the literal escape character, or
    readable forms like ``"ESC P"`` and ``"ESC x1_"``. Trailing CR/LF is
    stripped because Sartorius documents command terminators as optional and
    the existing fake-transport fixtures use the bare token.
    """
    if isinstance(command, bytes):
        token = command
    else:
        text = command.strip()
        if text.upper().startswith("ESC"):
            text = "\x1b" + text[3:].lstrip()
        token = text.encode("ascii", errors="strict")
    token = token.rstrip(b"\r\n")
    if not token:
        raise SartoriusValidationError(
            "SBI command token cannot be empty",
            context=ErrorContext(protocol="sbi"),
        )
    if not token.startswith(ESC):
        raise SartoriusValidationError(
            "SBI command token must start with ESC",
            context=ErrorContext(
                protocol="sbi",
                sbi_token=token,
                extra={"token": token.hex()},
            ),
        )
    return token

def split_lines(raw: bytes) -> tuple[bytes, ...]:
    r"""Split a raw SBI payload into complete lines.

    Every non-empty line must include ``\r\n`` (or at least ``\n`` for
    lenient fixture parsing). Returns lines including their terminators.
    """
    if raw == b"":
        return ()
    out: list[bytes] = []
    start = 0
    while start < len(raw):
        idx = raw.find(b"\n", start)
        if idx < 0:
            raise SartoriusFrameError(
                "SBI payload ended before a line terminator",
                context=ErrorContext(raw_response=raw, protocol="sbi"),
            )
        out.append(raw[start : idx + 1])
        start = idx + 1
    return tuple(out)