alicatlib.protocol¶

def parse_fields(
    raw: str,
    *,
    command: str,
    expected_count: int | None = None,
) -> list[str]:
    """Split a whitespace-delimited response into fields.

    Args:
        raw: Response text (already ASCII-decoded).
        command: Command name, for the error message.
        expected_count: If given, enforce exactly this many fields.

    Returns:
        The list of non-empty fields.

    Raises:
        AlicatParseError: If ``expected_count`` is set and the actual count
            differs.
    """
    fields = raw.split()
    if expected_count is not None and len(fields) != expected_count:
        raise AlicatParseError(
            f"{command}: expected {expected_count} fields, got {len(fields)} — {raw!r}",
            field_name="fields",
            expected=expected_count,
            actual=len(fields),
            context=ErrorContext(command_name=command, raw_response=raw.encode("ascii", "replace")),
        )
    return fields

def parse_float(value: str, *, field: str) -> float:
    """Parse ``value`` as a float, raising on failure."""
    try:
        return float(value)
    except ValueError as exc:
        raise AlicatParseError(
            f"could not parse float from {value!r} (field={field})",
            field_name=field,
            expected="float",
            actual=value,
        ) from exc

def parse_int(value: str, *, field: str) -> int:
    """Parse ``value`` as a base-10 integer, raising on failure."""
    try:
        return int(value)
    except ValueError as exc:
        raise AlicatParseError(
            f"could not parse integer from {value!r} (field={field})",
            field_name=field,
            expected="integer",
            actual=value,
        ) from exc

def strip_eol(data: bytes, *, eol: bytes = EOL) -> bytes:
    """Return ``data`` without a trailing ``eol``.

    Idempotent: if ``data`` already lacks the EOL, it is returned unchanged.
    """
    if data.endswith(eol):
        return data[: -len(eol)]
    return data

def parse_bool_code(
    value: str,
    *,
    field: str,
    mapping: Mapping[str, bool] = _DEFAULT_BOOL_MAPPING,
) -> bool:
    """Parse a boolean-coded field (default: ``"1"`` → True, ``"0"`` → False).

    Args:
        value: The wire-level field.
        field: Field name, for the error message.
        mapping: Accepted string-to-bool pairs. Override for commands that use
            non-standard codes (e.g. some commands use ``"Y"`` / ``"N"``).

    Raises:
        AlicatParseError: If ``value`` is not a key in ``mapping``.
    """
    try:
        return mapping[value]
    except KeyError as exc:
        accepted = ", ".join(repr(k) for k in mapping)
        raise AlicatParseError(
            f"could not parse bool from {value!r} (field={field}, accepted={accepted})",
            field_name=field,
            expected=tuple(mapping),
            actual=value,
        ) from exc

def parse_data_frame(raw: bytes, fmt: DataFrameFormat) -> ParsedFrame:
    """Parse ``raw`` against ``fmt`` into a :class:`ParsedFrame`.

    Thin delegator to :meth:`DataFrameFormat.parse`. Provided as a
    free-function alias so all low-level parsers share one import site
    (:mod:`alicatlib.protocol.parser`), matching the rest of design §5.11.
    Pure — no clocks; the session captures ``received_at`` /
    ``monotonic_ns`` and wraps via :meth:`DataFrame.from_parsed`.
    """
    return fmt.parse(raw)

def parse_data_frame_table(lines: Sequence[bytes]) -> DataFrameFormat:
    """Parse a ``??D*`` response into a :class:`DataFrameFormat`.

    Auto-detects the dialect by sniffing the column-header row, then
    dispatches to the appropriate per-dialect parser:

    - **V8+ dialect (DEFAULT)** — V8/V9 + V10 captures (design §16.6).
      Header: ``<uid> D00 ID_ NAME... TYPE... WIDTH NOTES...``.
      Field rows carry a stat-code column and conditional rows are
      marked with a leading ``*`` on the name.
    - **V1_V7 dialect** — 5v12 capture (design §16.6.2). Header:
      ``<uid>  D00 NAME... TYPE... MinVal MaxVal UNITS...``. No
      stat-code column, no ``*`` marker; engineering units sit in the
      trailing column.

    Per-field :attr:`DataFrameField.unit` is bound inline when the
    dialect carries a recognisable unit label.

    Raises:
        AlicatParseError: Non-ASCII bytes, or no field lines were
            recognised in either dialect.
    """
    flavor = _df_detect_flavor(lines)
    if flavor is DataFrameFormatFlavor.LEGACY:
        return _parse_data_frame_table_v1_v7(lines)
    return _parse_data_frame_table_default(lines)

def parse_enum_code[E: "Gas | Statistic | Unit"](
    value: str,
    *,
    field: str,
    registry: AliasRegistry[E],
) -> E:
    """Parse a numeric code and resolve it against ``registry``.

    Wraps :meth:`alicatlib.registry.aliases.AliasRegistry.by_code` so that
    unknown codes coming from the device surface as :class:`AlicatParseError`
    (a protocol-layer problem) rather than :class:`UnknownGasError` /
    :class:`UnknownStatisticError` (config-layer problems from user input).
    The original registry error is preserved as ``__cause__``.

    Only usable with :class:`AliasRegistry` (unique-code enums: Gas, Statistic).
    Unit lookups require a :class:`UnitCategory` disambiguator and are handled
    at the data-frame-field parsing layer where the category is known.
    """
    code = parse_int(value, field=field)
    try:
        return registry.by_code(code)
    except (UnknownGasError, UnknownStatisticError, UnknownUnitError) as exc:
        raise AlicatParseError(
            f"unknown code {code} for {field}",
            field_name=field,
            expected="known registry code",
            actual=code,
        ) from exc

def parse_fields(
    raw: str,
    *,
    command: str,
    expected_count: int | None = None,
) -> list[str]:
    """Split a whitespace-delimited response into fields.

    Args:
        raw: Response text (already ASCII-decoded).
        command: Command name, for the error message.
        expected_count: If given, enforce exactly this many fields.

    Returns:
        The list of non-empty fields.

    Raises:
        AlicatParseError: If ``expected_count`` is set and the actual count
            differs.
    """
    fields = raw.split()
    if expected_count is not None and len(fields) != expected_count:
        raise AlicatParseError(
            f"{command}: expected {expected_count} fields, got {len(fields)} — {raw!r}",
            field_name="fields",
            expected=expected_count,
            actual=len(fields),
            context=ErrorContext(command_name=command, raw_response=raw.encode("ascii", "replace")),
        )
    return fields

def parse_float(value: str, *, field: str) -> float:
    """Parse ``value`` as a float, raising on failure."""
    try:
        return float(value)
    except ValueError as exc:
        raise AlicatParseError(
            f"could not parse float from {value!r} (field={field})",
            field_name=field,
            expected="float",
            actual=value,
        ) from exc

def parse_gas_list(lines: Sequence[bytes]) -> dict[int, str]:
    """Parse a ``??G*`` response into ``{gas_code: short_name}``.

    Real V10 wire shape (verified 2026-04-17 on a MC-500SCCM-D, design
    §16.6)::

        <unit_id> G<NN>      <short_name>

    The integer in the ``G<NN>`` row label is the device-side gas code
    (which coincides with the canonical Appendix-C code for the built-in
    gases G00..G29 and continues with mixture/specialty slots beyond).
    The short name (e.g. ``Air``, ``CH4``, ``N2``) is right-aligned in a
    fixed-width column; we collapse the leading whitespace.

    Invariants:

    - A consistent ``unit_id`` across every parsed line — mismatch
      raises :class:`AlicatUnitIdMismatchError`.
    - Duplicate gas codes raise :class:`AlicatParseError` rather than
      silently overwriting — a duplicate would mask firmware oddities.
    - Empty responses or responses with no recognisable gas lines raise
      :class:`AlicatParseError`; the ``??G*`` command always has at
      least one built-in gas on any supported device.

    Returns:
        Mapping from Alicat gas code (per-device, often matching primer
        Appendix C for built-ins) to the wire short name.
        :meth:`gas_registry.coerce` resolves the short name to the typed
        :class:`~alicatlib.registry.Gas` member.
    """
    if not lines:
        raise AlicatParseError(
            "??G*: empty response",
            field_name="gas_list",
            expected=">=1 line",
            actual=0,
            context=ErrorContext(command_name="??G*"),
        )

    # GP firmware returns the full list on one line; split it before the
    # per-line loop so every downstream invariant stays the same.
    lines = _maybe_split_gp_inline_gas_list(lines)

    unit_id: str | None = None
    by_code: dict[int, str] = {}

    for raw_line in lines:
        text = _strip_gp_padding(parse_ascii(raw_line).rstrip("\r\n"))
        if not text.strip():
            continue
        m = _GAS_LIST_LINE_RE.match(text)
        if m is None:
            # Lines that don't match the row shape (preamble, blank, etc.)
            # are silently skipped.
            continue

        label = m.group("label").strip()
        if not label:
            continue

        line_uid = m.group("uid")
        code = int(m.group("code"))

        if unit_id is None:
            unit_id = line_uid
        elif unit_id != line_uid:
            raise AlicatUnitIdMismatchError(
                f"??G*: unit_id {line_uid!r} on G{m.group('code')} line does not match {unit_id!r}",
                context=ErrorContext(
                    command_name="??G*",
                    unit_id=unit_id,
                    raw_response=raw_line,
                ),
            )

        if code in by_code:
            raise AlicatParseError(
                f"??G*: duplicate gas code {code} — {text!r}",
                field_name="gas_list_code",
                expected="unique gas codes",
                actual=code,
                context=ErrorContext(command_name="??G*", raw_response=raw_line),
            )

        by_code[code] = label

    if not by_code:
        raise AlicatParseError(
            "??G*: no gas entries recognised in response",
            field_name="gas_list",
            expected=">=1 gas entry",
            actual=0,
            context=ErrorContext(command_name="??G*"),
        )

    return by_code

def parse_int(value: str, *, field: str) -> int:
    """Parse ``value`` as a base-10 integer, raising on failure."""
    try:
        return int(value)
    except ValueError as exc:
        raise AlicatParseError(
            f"could not parse integer from {value!r} (field={field})",
            field_name=field,
            expected="integer",
            actual=value,
        ) from exc

def parse_manufacturing_info(lines: Sequence[bytes]) -> ManufacturingInfo:
    """Parse a ``??M*`` response into :class:`ManufacturingInfo`.

    Expected shape (per primer — verified against hardware fixtures as
    they're captured): a series of ``<unit_id> M<NN> <payload>`` lines,
    one per code. The parser does not pin a line count (firmware versions
    have varied) but does enforce:

    - A consistent ``unit_id`` across every line — a mismatch is an
      :class:`AlicatUnitIdMismatchError` (a sign the bus has bled frames
      from another device).
    - No duplicate ``M<NN>`` codes within the response — a duplicate is
      an :class:`AlicatParseError` rather than a silent overwrite.
    - Every non-empty line matches the ``<uid> M<NN> <payload>`` shape;
      a malformed line raises rather than being skipped, so firmware-side
      format drift surfaces instead of being swallowed.

    The *semantic* mapping (``M04`` → model, ``M05`` → serial, etc.) is
    intentionally not applied here — it belongs in the factory
    (:mod:`alicatlib.devices.factory`) where it can be firmware-version
    aware and validated against real captures.
    """
    if not lines:
        raise AlicatParseError(
            "??M*: empty response",
            field_name="manufacturing_info",
            expected=">=1 line",
            actual=0,
            context=ErrorContext(command_name="??M*"),
        )

    unit_id: str | None = None
    by_code: dict[int, str] = {}

    for raw_line in lines:
        # Strip GP backspace-padding before regex match — see _strip_gp_padding.
        text = _strip_gp_padding(parse_ascii(raw_line).rstrip("\r\n"))
        if not text:
            continue
        m = _MFG_LINE_RE.match(text)
        if m is None:
            raise AlicatParseError(
                f"??M*: malformed line {text!r}",
                field_name="manufacturing_info_line",
                expected="<unit_id> M<NN> <payload>",
                actual=text,
                context=ErrorContext(command_name="??M*", raw_response=raw_line),
            )

        line_uid = m.group("uid")
        code = int(m.group("code"))
        payload = (m.group("payload") or "").strip()

        if unit_id is None:
            unit_id = line_uid
        elif unit_id != line_uid:
            raise AlicatUnitIdMismatchError(
                f"??M*: unit_id {line_uid!r} on M{code:02d} line does not match {unit_id!r}",
                context=ErrorContext(
                    command_name="??M*",
                    unit_id=unit_id,
                    raw_response=raw_line,
                ),
            )

        if code in by_code:
            raise AlicatParseError(
                f"??M*: duplicate M{code:02d} line — {text!r}",
                field_name="manufacturing_info_code",
                expected="unique M-codes",
                actual=code,
                context=ErrorContext(command_name="??M*", raw_response=raw_line),
            )

        by_code[code] = payload

    if unit_id is None:
        # All lines were blank — treat as empty response.
        raise AlicatParseError(
            "??M*: response contained only blank lines",
            field_name="manufacturing_info",
            expected=">=1 non-blank line",
            actual=0,
            context=ErrorContext(command_name="??M*"),
        )

    return ManufacturingInfo(unit_id=unit_id, by_code=MappingProxyType(by_code))

def parse_optional_float(value: str, *, field: str) -> float | None:
    """Parse ``value`` as a float, returning ``None`` for the ``"--"`` sentinel.

    Alicat emits ``--`` in a data frame when a field is unavailable on the
    current device or in the current mode (e.g. setpoint on a flow meter).
    Callers that want a strict parse should use :func:`parse_float` directly.
    """
    if value == _ABSENT_TOKEN:
        return None
    return parse_float(value, field=field)

def parse_status_codes(tokens: Sequence[str]) -> frozenset[StatusCode]:
    """Collect :class:`StatusCode` members from a token sequence.

    Any token whose value is not a known status code is silently skipped —
    callers that want "status-only" semantics should pre-slice the tail of
    their token stream, since status codes are the trailing run of a data
    frame per primer convention.

    Order on the wire is not preserved (returned as a :class:`frozenset`);
    the primer does not specify a canonical ordering for multi-code runs.
    """
    return frozenset(StatusCode(t) for t in tokens if t in _STATUS_VALUES)

def parse_ve_response(raw: bytes) -> tuple[FirmwareVersion, date | None]:
    """Parse a ``VE`` (firmware version) response.

    ``VE`` is the one identification command that works on every firmware
    family — it is the anchor of the identification pipeline (design §5.9).
    The response shape varies across families: at minimum it contains a
    firmware token (``GP``, ``GP-10v05``, ``10v05``, ``1v00``, ...); some
    devices additionally report a firmware date in ISO ``YYYY-MM-DD`` form.

    This parser is deliberately tolerant: it scans the decoded response for
    (a) the first firmware-shaped token and (b) the first ISO-date token,
    regardless of their relative position or surrounding text.

    Returns:
        A ``(FirmwareVersion, date | None)`` pair. The date is ``None`` when
        the device's firmware does not include one.

    Raises:
        AlicatParseError: If no firmware token can be found. A malformed date
            (present but unparseable) also raises, rather than silently
            dropping the field — a garbled date is a sign of line corruption
            that the caller should see.
    """
    text = parse_ascii(raw)
    fw_match = _FIRMWARE_TOKEN_RE.search(text)
    if fw_match is None:
        raise AlicatParseError(
            f"VE: no firmware token in response {text!r}",
            field_name="firmware",
            expected="GP or <major>v<minor>",
            actual=text,
            context=ErrorContext(command_name="VE", raw_response=raw),
        )

    fw = FirmwareVersion.parse(fw_match.group(0))

    iso_match = _ISO_DATE_RE.search(text)
    if iso_match is not None:
        try:
            fw_date = date(
                int(iso_match.group("y")),
                int(iso_match.group("m")),
                int(iso_match.group("d")),
            )
        except ValueError as exc:
            raise AlicatParseError(
                f"VE: malformed date {iso_match.group(0)!r} in response {text!r}",
                field_name="firmware_date",
                expected="YYYY-MM-DD",
                actual=iso_match.group(0),
                context=ErrorContext(command_name="VE", raw_response=raw),
            ) from exc
        return fw, fw_date

    name_match = _MONTH_NAME_DATE_RE.search(text)
    if name_match is not None:
        try:
            fw_date = date(
                int(name_match.group("y")),
                _MONTH_TO_NUM[name_match.group("m").lower()],
                int(name_match.group("d")),
            )
        except ValueError as exc:
            raise AlicatParseError(
                f"VE: malformed date {name_match.group(0)!r} in response {text!r}",
                field_name="firmware_date",
                expected="<Mon> <DD> <YYYY>",
                actual=name_match.group(0),
                context=ErrorContext(command_name="VE", raw_response=raw),
            ) from exc
        return fw, fw_date

    return fw, None

def decode_ascii(raw: bytes) -> str:
    """Decode ``raw`` as ASCII, raising :class:`AlicatParseError` on non-ASCII.

    The Alicat wire format is ASCII-only — non-ASCII bytes indicate line
    noise or a framing error, not a legitimate extended-charset response,
    so raising with the raw bytes preserved is the right behaviour.
    Re-exported as :func:`alicatlib.protocol.parser.parse_ascii` for
    callers that already import from the parser module; implementation
    lives here so :mod:`alicatlib.devices.data_frame` can use it without
    introducing a parser-layer import cycle.
    """
    try:
        return raw.decode("ascii")
    except UnicodeDecodeError as exc:
        raise AlicatParseError(
            f"non-ASCII bytes in response: {raw!r}",
            field_name="response",
            expected="ASCII bytes",
            actual=raw,
            context=ErrorContext(raw_response=raw),
        ) from exc

def strip_eol(data: bytes, *, eol: bytes = EOL) -> bytes:
    """Return ``data`` without a trailing ``eol``.

    Idempotent: if ``data`` already lacks the EOL, it is returned unchanged.
    """
    if data.endswith(eol):
        return data[: -len(eol)]
    return data