sartoriuslib.testing¶

def build_identify_script(
    *,
    model: str = "MSE1203S-100-DR",
    manufacturer: str = "Sartorius",
    software: bytes | None = None,
    factory_number: bytes | None = None,
    sbn: int = 0x00,
) -> dict[bytes, bytes]:
    """Build a scripted transport mapping for :meth:`Balance.identify`.

    The returned dict covers every TX frame the identify sequence
    sends (model, manufacturer, software, factory number, SBN) so a
    :class:`FakeTransport` constructed with it can drive
    :func:`open_device` from end to end without hardware.
    """
    if software is None:
        software = bytes([0x00, 0x39, 0x21, 0x00, 0x39, 0x01, 0x39, 0x01, 0x00, 0x01])
    if factory_number is None:
        factory_number = bytes([0x00, 0x31, 0x80, 0x11, 0x65])

    return {
        canned_frames.TX_READ_MODEL: _ascii_blob(0x54, 20, model),
        canned_frames.TX_READ_MANUFACTURER: _ascii_blob(0x50, 16, manufacturer),
        canned_frames.TX_READ_SW_VERSION: _rx(0x4A, software),
        canned_frames.TX_READ_FACTORY_NUMBER: _rx(0x45, factory_number),
        canned_frames.TX_READ_SBN: _short_data_u8(sbn),
    }

def build_metrology_script(
    *,
    capacity_g: float = 1200.0,
    increment_g: float = 0.001,
    config_counter: int | None = 1,
    area: int = 0,
) -> dict[bytes, bytes]:
    """Build a ``{tx: rx}`` script covering capacity / increment / counter.

    Pair with :func:`build_identify_script` to script a full
    open sequence (identity + metrology probe). Defaults match the
    MSE1203S unit in ``docs/protocol.md``.

    Pass ``config_counter=None`` to omit the ``0xBA`` reply entirely —
    use that when simulating a balance that genuinely lacks
    :attr:`Capability.CONFIG_COUNTER` (e.g. WZA). The new probe-driven
    capability detection in :meth:`Balance._probe_dispatch_capabilities`
    relies on a missing reply (not a falsy value) to decide the cap is
    absent, so unfaithful "always reply" fixtures would silently
    over-report the capability.
    """
    tx_capacity = build_command(0x0C, bytes([0x21, area]))
    tx_increment = build_command(0x0D, bytes([0x21, area]))
    script: dict[bytes, bytes] = {
        tx_capacity: _typed_float_rx(capacity_g),
        tx_increment: _typed_float_rx(increment_g),
    }
    if config_counter is not None:
        tx_counter = build_command(0xBA)
        script[tx_counter] = _short_data_u8(config_counter)
    return script

def build_parameter_read_script(
    index: int,
    current: int,
    max_value: int,
) -> dict[bytes, bytes]:
    """Build a one-entry script for a ``read_parameter(index)`` call."""
    tx = build_command(0x55, bytes([0x21, index]))
    return {tx: _parameter_rx(current, max_value)}

def build_parameter_write_script(index: int, value: int) -> dict[bytes, bytes]:
    """Build a one-entry script for a ``write_parameter(index, value)`` call."""
    tx = build_command(0x56, bytes([0x21, index, 0x21, value]))
    return {tx: canned_frames.RX_ACK}

def build_sbi_identify_script(
    *,
    model: str = "WZA8202-N",
    serial: str = "12345678",
    software: str = "1.0",
) -> dict[bytes, bytes]:
    """Build a scripted SBI identity mapping for ``open_device(..., SBI)``."""
    return {
        TOKEN_TYPE: _sbi_line(model),
        TOKEN_SERIAL: _sbi_line(serial),
        TOKEN_SOFTWARE: _sbi_line(software),
    }

def build_temperature_script(
    *,
    sensor_celsius: dict[int, float | None],
    out_of_range_after: int | None = None,
) -> dict[bytes, bytes]:
    """Script the per-sensor replies for ``temperature(N)`` calls.

    ``sensor_celsius`` maps sensor index → temperature in °C, or ``None``
    to script the ``7f ff ff ff`` "reserved slot" sentinel for that
    index. ``out_of_range_after`` adds an xBPI ``0x04`` (unknown opcode)
    reply for index ``N`` (and the helper does NOT script higher
    indices, mirroring the wire reality where the device stops
    replying past the end). Useful for testing
    :meth:`Balance.discover_temperature_sensors` against a sparse
    layout (e.g. MSE: ``{0: 25.5, 1: 25.6, 2: None, 3: 36.7}`` +
    ``out_of_range_after=4``).
    """
    sentinel_body = b"\x7f\xff\xff\xff\xff"
    script: dict[bytes, bytes] = {}
    for sensor, celsius in sensor_celsius.items():
        tx = build_command(0x76, bytes([0x21, sensor]))
        if celsius is None:
            script[tx] = _rx(0x35, sentinel_body)
        else:
            script[tx] = _typed_float_rx(celsius)
    if out_of_range_after is not None:
        tx = build_command(0x76, bytes([0x21, out_of_range_after]))
        # xBPI 0x01 0x04 = unknown/unsupported opcode.
        script[tx] = _rx(0x01, b"\x04")
    return script

def parse_sbi_fixture(text: str) -> dict[bytes, bytes]:
    r"""Parse an SBI text fixture into a ``{tx_bytes: rx_bytes}`` mapping.

    Format::

        # SBI fixture: print
        > ESC P
        < +     0.00 g

    Reply lines get ``\r\n`` appended when the fixture omits a terminator.
    Multiple ``<`` lines after one ``>`` concatenate into a multi-line reply.
    """
    mapping: dict[bytes, bytes] = {}
    current_tx: bytes | None = None
    for lineno, raw in enumerate(text.splitlines(), start=1):
        line = raw.split("#", 1)[0].strip()
        if not line:
            continue
        marker, _, rest = line.partition(" ")
        payload = rest.strip()
        if marker == ">":
            if not payload:
                raise SartoriusValidationError(
                    f"fixture line {lineno}: empty SBI TX payload",
                    context=ErrorContext(extra={"line": lineno}),
                )
            try:
                current_tx = normalize_token(payload)
            except Exception as exc:
                raise SartoriusValidationError(
                    f"fixture line {lineno}: invalid SBI token {payload!r}",
                    context=ErrorContext(extra={"line": lineno, "payload": payload}),
                ) from exc
            mapping.setdefault(current_tx, b"")
        elif marker == "<":
            if current_tx is None:
                raise SartoriusValidationError(
                    f"fixture line {lineno}: '<' reply before any '>' request",
                    context=ErrorContext(extra={"line": lineno}),
                )
            mapping[current_tx] = mapping[current_tx] + _sbi_line(payload)
        else:
            raise SartoriusValidationError(
                f"fixture line {lineno}: unrecognised marker {marker!r} (expected '>' or '<')",
                context=ErrorContext(extra={"line": lineno, "marker": marker}),
            )
    return mapping

def parse_xbpi_fixture(text: str) -> dict[bytes, bytes]:
    r"""Parse an xBPI text fixture into a ``{tx_bytes: rx_bytes}`` mapping.

    Format (design §8.2)::

        # xBPI fixture: read net weight
        > 04 01 09 1e 2c
        < 0b 41 48 bb a3 d7 0a 3d 30 82 45 07

    - Blank lines and ``#``-comment lines are ignored.
    - ``>`` lines carry TX bytes (host→balance).
    - ``<`` lines carry RX bytes (balance→host); they attach to the most
      recent ``>`` line.
    - Hex digits may be separated by whitespace in any form.
    - Multiple ``<`` lines following one ``>`` concatenate into one
      reply blob (useful for synthesising multi-frame responses).

    Raises :class:`SartoriusValidationError` for malformed input (an
    ``<`` line before any ``>``, odd-length hex tokens, etc.).
    """
    mapping: dict[bytes, bytes] = {}
    current_tx: bytes | None = None
    for lineno, raw in enumerate(text.splitlines(), start=1):
        line = raw.split("#", 1)[0].strip()
        if not line:
            continue
        marker, _, rest = line.partition(" ")
        payload = rest.strip()
        if marker == ">":
            if not payload:
                raise SartoriusValidationError(
                    f"fixture line {lineno}: empty TX payload",
                    context=ErrorContext(extra={"line": lineno}),
                )
            current_tx = _decode_hex_tokens(payload, lineno)
            mapping.setdefault(current_tx, b"")
        elif marker == "<":
            if current_tx is None:
                raise SartoriusValidationError(
                    f"fixture line {lineno}: '<' reply before any '>' request",
                    context=ErrorContext(extra={"line": lineno}),
                )
            rx = _decode_hex_tokens(payload, lineno) if payload else b""
            mapping[current_tx] = mapping[current_tx] + rx
        else:
            raise SartoriusValidationError(
                f"fixture line {lineno}: unrecognised marker {marker!r} (expected '>' or '<')",
                context=ErrorContext(extra={"line": lineno, "marker": marker}),
            )
    return mapping