fujilib.testing¶

def hex_to_bytes(text: str) -> bytes:
    """Parse hex bytes separated by spaces, colons or commas.

    Raises:
        FujiValidationError: a token is not one or two hex digits, or ``text`` is empty.
    """
    tokens = text.replace(":", " ").replace(",", " ").split()
    if not tokens:
        msg = "no hex bytes"
        raise FujiValidationError(msg)
    out = bytearray()
    for token in tokens:
        if len(token) > 2:  # noqa: PLR2004
            if len(token) % 2:
                msg = f"odd-length hex {token!r}"
                raise FujiValidationError(msg)
            pairs = [token[i : i + 2] for i in range(0, len(token), 2)]
        else:
            pairs = [token]
        for pair in pairs:
            try:
                out.append(int(pair, 16))
            except ValueError:
                msg = f"{pair!r} is not a hex byte"
                raise FujiValidationError(msg) from None
    return bytes(out)

def parse_arrow_fixture(source: str | Path, *, name: str | None = None) -> tuple[Exchange, ...]:
    """Parse an arrow fixture from a path or from its text.

    Args:
        source: A :class:`~pathlib.Path` to read, or the fixture text itself.
        name: The label for error messages; defaults to the path, or ``<string>``.

    Raises:
        FujiValidationError: a malformed line; the message starts with ``name:line``.
    """
    if isinstance(source, Path):
        label = name or str(source)
        text = source.read_text(encoding="utf-8")
    else:
        label = name or "<string>"
        text = source
    exchanges: list[Exchange] = []
    request: tuple[bytes, int, str] | None = None
    response: bytearray | None = None

    def flush() -> None:
        if request is not None:
            data, line, comment = request
            reply = bytes(response) if response is not None else None
            exchanges.append(Exchange(data, reply, line, comment))

    for number, raw in enumerate(text.splitlines(), start=1):
        content, _, comment = raw.partition("#")
        content = content.strip()
        if not content:
            continue
        marker, _, payload = content.partition(" ")
        try:
            data = hex_to_bytes(payload)
        except FujiValidationError as exc:
            msg = f"{label}:{number}: {exc}"
            raise FujiValidationError(msg, context=ErrorContext(extra={"line": number})) from None
        if marker == ">":
            flush()
            request, response = (data, number, comment.strip()), None
        elif marker == "<":
            if request is None:
                msg = f"{label}:{number}: a reply before any request"
                raise FujiValidationError(msg, context=ErrorContext(extra={"line": number}))
            response = (response or bytearray()) + data
        else:
            msg = f"{label}:{number}: a line must start with '>' or '<', not {marker!r}"
            raise FujiValidationError(msg, context=ErrorContext(extra={"line": number}))
    flush()
    return tuple(exchanges)

def replay_script(exchanges: tuple[Exchange, ...]) -> Mapping[bytes, bytes]:
    """Map each request to its reply, for replaying a fixture to a client.

    Requests without a reply are left out.

    Raises:
        FujiValidationError: the same request appears with two different replies.
    """
    script: dict[bytes, bytes] = {}
    for exchange in exchanges:
        if exchange.response is None:
            continue
        known = script.get(exchange.request)
        if known is not None and known != exchange.response:
            msg = f"line {exchange.line}: this request already has a different reply"
            raise FujiValidationError(msg, context=ErrorContext(extra={"line": exchange.line}))
        script[exchange.request] = exchange.response
    return MappingProxyType(script)

def analyzer(
    *,
    instrument_error: bool = False,
    errors: tuple[ErrorCode, ...] = (),
    alarms: tuple[AlarmState | int, ...] = (AlarmState.NONE,) * 6,
    auto_calibration: bool = False,
) -> AnalyzerStatus:
    """The analyzer-level status, on the measurement screen with nothing active.

    Args:
        instrument_error: The instrument-error flag.
        errors: The analyzer's active errors: 1, 2, 3 or 10.
        alarms: The states of alarms 1-6, in order.
        auto_calibration: An auto calibration is running.
    """
    return AnalyzerStatus(
        instrument_error=instrument_error,
        calibration_error=False,
        errors=frozenset(errors),
        alarms=alarms,
        peak_count=0,
        peak_alarm=False,
        auto_calibration_running=auto_calibration,
        display=DisplayState(
            screen=DisplayScreen.MEASUREMENT,
            calibration_step=ManualCalibrationStep.NONE,
            top_channel=ChannelId.CH1,
            cursor_channel=None,
        ),
    )

def bench_readings() -> tuple[Reading, ...]:
    """CO2, CO and O2 as the bench ZPA reported them at capture."""
    return (
        reading(ChannelId.CH1, Gas.CO2, -11, 2),
        reading(ChannelId.CH2, Gas.CO, -9, 3),
        reading(ChannelId.CH3, Gas.O2, 2029, 2),
    )

def frame(
    readings: tuple[Reading, ...] | None = None,
    status_block: AnalyzerStatus | None = None,
    *,
    detail: bool = True,
    readings_timing: TransferTiming | None = None,
    status_timing: TransferTiming | None = None,
) -> Frame:
    """A poll's frame.

    Args:
        readings: The channels; :func:`bench_readings` if omitted.
        status_block: The analyzer status; :func:`analyzer`'s default if omitted.
        detail: ``False`` builds the frame of a poll that read only the
            concentration block: no analyzer status and no status timing.
        readings_timing: The concentration block's timing, which times the
            sample; ``timing(0.0)`` if omitted.
        status_timing: The status block's timing; ``timing(25.0)`` if omitted.
            Ignored without ``detail``.
    """
    if status_timing is None:
        status_timing = timing(25.0)
    return Frame(
        readings=readings if readings is not None else bench_readings(),
        analyzer=(status_block or analyzer()) if detail else None,
        protocol=ProtocolKind.MODBUS_RTU,
        readings_timing=readings_timing if readings_timing is not None else timing(0.0),
        status_timing=status_timing if detail else None,
        raw=b"\x00" * _POLL_BYTES,
    )

def reading(
    channel: ChannelId,
    gas: Gas,
    raw: int,
    decimals: int,
    *,
    unit: Unit = Unit.VOL_PERCENT,
    state: ReadingState = ReadingState.OK,
    channel_status: ChannelStatus | None = None,
    label_source: LabelSource = LabelSource.ASSERTED,
    role: ChannelRole = ChannelRole.INSTANTANEOUS,
) -> Reading:
    """One channel's reading, its value ``raw / 10**decimals``.

    Args:
        channel: The channel.
        gas: The gas it carries, also given as the suggested gas.
        raw: The signed integer of the concentration register.
        decimals: The decimal-point register, 0-3.
        unit: The unit of the channel's current range.
        state: The validity state; it is not derived from ``channel_status``.
        channel_status: The channel's status; :func:`status`'s default if omitted.
        label_source: Where the gas label comes from.
        role: The channel's role.
    """
    return Reading(
        channel=channel,
        gas=gas,
        suggested_gas=gas,
        label_source=label_source,
        role=role,
        value=raw / 10**decimals,
        unit=unit,
        raw_value=raw,
        decimals=decimals,
        status=channel_status if channel_status is not None else status(),
        state=state,
        protocol=ProtocolKind.MODBUS_RTU,
    )

def status(
    *,
    rng: int = 1,
    hold: bool = False,
    zero: bool = False,
    span: bool = False,
    errors: tuple[ErrorCode, ...] = (),
) -> ChannelStatus:
    """A measured channel's status: live on range 1 unless told otherwise.

    Args:
        rng: The current range, 1 or 2.
        hold: The channel's output is held.
        zero: A manual zero calibration of the channel is under way.
        span: A manual span calibration of the channel is under way.
        errors: The channel's active errors, 4-9.
    """
    return ChannelStatus(
        range=rng,
        zero_calibrating=zero,
        span_calibrating=span,
        auto_zero_running=False,
        auto_span_running=False,
        hold=hold,
        errors=frozenset(errors),
    )

def timing(
    offset_ms: float = 0.0,
    latency_ms: float = 20.0,
    *,
    origin: datetime = T0,
    mono_origin_ns: int = MONO0,
) -> TransferTiming:
    """The timing of a transaction sent ``offset_ms`` after the origin.

    Args:
        offset_ms: When the request was sent, in milliseconds after the origin.
        latency_ms: The round trip, request to reply.
        origin: The wall-clock origin, tz-aware.
        mono_origin_ns: The monotonic origin, the same instant as ``origin``.
    """
    start = origin + timedelta(milliseconds=offset_ms)
    mono = mono_origin_ns + int(offset_ms * 1e6)
    return TransferTiming(
        requested_at=start,
        received_at=start + timedelta(milliseconds=latency_ms),
        t_request_mono_ns=mono,
        t_reply_mono_ns=mono + int(latency_ms * 1e6),
    )