alicatlib.commands¶

def decode(
    self,
    response: bytes | tuple[bytes, ...],
    ctx: DecodeContext,
) -> AnalogOutputSourceSetting:
    """Parse 4-field reply into :class:`AnalogOutputSourceSetting`.

    ``channel`` is request-echo; the device doesn't re-echo it so
    the facade fills from the request.
    """
    del ctx
    if isinstance(response, tuple):
        raise TypeError(
            f"{self.name}.decode expected single-line response, got {len(response)} lines",
        )
    text = response.decode("ascii")
    fields = parse_fields(text, command=self.name, expected_count=4)
    unit_id, value_s, code_s, label = fields
    value = parse_int(value_s, field=f"{self.name}.value")
    code = parse_int(code_s, field=f"{self.name}.unit_code")
    return AnalogOutputSourceSetting(
        unit_id=unit_id,
        channel=AnalogOutputChannel.PRIMARY,  # facade replaces with request.channel
        value=value,
        unit_code=code,
        unit=_resolve_unit_label(label, code),
        unit_label=label,
    )

def encode(self, ctx: DecodeContext, request: AnalogOutputSourceRequest) -> bytes:
    """Emit ASOCV query or set bytes."""
    if request.value is not None and request.value < 0:
        raise AlicatValidationError(
            f"{self.name}: value must be non-negative (0/1 are min/max "
            f"sentinels; ≥2 are statistic codes); got {request.value}",
            context=ErrorContext(
                command_name=self.name,
                unit_id=ctx.unit_id,
                extra={"value": request.value},
            ),
        )
    prefix = ctx.command_prefix.decode("ascii")
    head = f"{ctx.unit_id}{prefix}{self.token} {int(request.channel)}"
    if request.value is None:
        return (head + "\r").encode("ascii")
    tokens = [head, str(request.value)]
    if request.unit_code is not None:
        tokens.append(str(request.unit_code))
    return (" ".join(tokens) + "\r").encode("ascii")

def decode(
    self,
    response: bytes | tuple[bytes, ...],
    ctx: DecodeContext,
) -> AutoTareState:
    """Parse ``<uid> <enable> <delay>`` into :class:`AutoTareState`."""
    del ctx
    if isinstance(response, tuple):
        raise TypeError(
            f"{self.name}.decode expected single-line response, got {len(response)} lines",
        )
    text = response.decode("ascii")
    fields = parse_fields(text, command=self.name, expected_count=3)
    unit_id, enable_s, delay_s = fields
    return AutoTareState(
        unit_id=unit_id,
        enabled=parse_bool_code(enable_s, field=f"{self.name}.enabled"),
        delay_s=parse_float(delay_s, field=f"{self.name}.delay_s"),
    )

def encode(self, ctx: DecodeContext, request: AutoTareRequest) -> bytes:
    """Emit ZCA query or set bytes.

    Disable uses the shortest accepted form (``ZCA 0``) rather
    than the primer's ``ZCA 0 0`` — the latter rejects on real
    10v20 (see §16.6.10 + class docstring).
    """
    prefix = ctx.command_prefix.decode("ascii")
    head = f"{ctx.unit_id}{prefix}{self.token}"
    if request.enable is None:
        return f"{head}\r".encode("ascii")
    # Disable form — no delay field on the wire.
    if not request.enable:
        return f"{head} 0\r".encode("ascii")
    # Enable form — delay required.
    delay = request.delay_s
    if delay is None:
        raise AlicatValidationError(
            f"{self.name}: delay_s is required when enabling auto-tare "
            f"(range {ZCA_DELAY_MIN_S}..{ZCA_DELAY_MAX_S} seconds)",
            context=ErrorContext(
                command_name=self.name,
                unit_id=ctx.unit_id,
            ),
        )
    if delay < ZCA_DELAY_MIN_S or delay > ZCA_DELAY_MAX_S:
        raise AlicatValidationError(
            f"{self.name}: delay_s must be in "
            f"[{ZCA_DELAY_MIN_S}, {ZCA_DELAY_MAX_S}] seconds, got {delay}",
            context=ErrorContext(
                command_name=self.name,
                unit_id=ctx.unit_id,
                extra={"delay_s": delay},
            ),
        )
    return f"{head} 1 {delay}\r".encode("ascii")

def decode(
    self,
    response: bytes | tuple[bytes, ...],
    ctx: DecodeContext,
) -> AverageTimingSetting:
    """Parse the ``DCA`` reply into :class:`AverageTimingSetting`.

    Primer p. 18 documents ``<uid> <stat> <averaging_ms>`` (3
    fields). Hardware validation (2026-04-17) on 10v20 firmware shows
    the device omits the ``<stat>`` echo and replies with just
    ``<uid> <averaging_ms>`` (2 fields). Accept both shapes; when
    the 2-field form arrives, ``statistic_code`` is ``0`` and the
    facade re-populates it from the request.
    """
    del ctx
    if isinstance(response, tuple):
        raise TypeError(
            f"{self.name}.decode expected single-line response, got {len(response)} lines",
        )
    text = response.decode("ascii")
    fields = parse_fields(text, command=self.name)
    if len(fields) == _DCA_FIELDS_WITH_STAT:
        unit_id, stat_s, avg_s = fields
        statistic_code = parse_int(stat_s, field=f"{self.name}.statistic_code")
    elif len(fields) == _DCA_FIELDS_WITHOUT_STAT:
        unit_id, avg_s = fields
        statistic_code = 0
    else:
        raise AlicatParseError(
            f"{self.name}: expected 2 or 3 fields, got {len(fields)} — {text!r}",
            field_name="average_timing",
            expected="2 or 3 fields",
            actual=len(fields),
            context=ErrorContext(command_name=self.name, raw_response=response),
        )
    return AverageTimingSetting(
        unit_id=unit_id,
        statistic_code=statistic_code,
        averaging_ms=parse_int(avg_s, field=f"{self.name}.averaging_ms"),
    )

def encode(self, ctx: DecodeContext, request: AverageTimingRequest) -> bytes:
    """Emit the DCA query or set bytes."""
    if request.statistic_code not in DCA_ALLOWED_STATISTIC_CODES:
        raise AlicatValidationError(
            f"{self.name}: statistic_code {request.statistic_code} not in "
            f"{sorted(DCA_ALLOWED_STATISTIC_CODES)} — DCA only averages "
            "pressure / flow statistics per primer p. 18.",
            context=ErrorContext(
                command_name=self.name,
                unit_id=ctx.unit_id,
                extra={"statistic_code": request.statistic_code},
            ),
        )
    prefix = ctx.command_prefix.decode("ascii")
    head = f"{ctx.unit_id}{prefix}{self.token} {request.statistic_code}"
    if request.averaging_ms is None:
        return (head + "\r").encode("ascii")
    if request.averaging_ms < 0 or request.averaging_ms > _DCA_MAX_AVERAGE_MS:
        raise AlicatValidationError(
            f"{self.name}: averaging_ms must be in [0, {_DCA_MAX_AVERAGE_MS}], "
            f"got {request.averaging_ms}",
            context=ErrorContext(
                command_name=self.name,
                unit_id=ctx.unit_id,
                extra={"averaging_ms": request.averaging_ms},
            ),
        )
    return f"{head} {request.averaging_ms}\r".encode("ascii")

def decode(
    self,
    response: bytes | tuple[bytes, ...],
    ctx: DecodeContext,
) -> BlinkDisplayState:
    """Parse ``<uid> <0|1>`` into :class:`BlinkDisplayState`."""
    del ctx
    if isinstance(response, tuple):
        raise TypeError(
            f"{self.name}.decode expected single-line response, got {len(response)} lines",
        )
    text = response.decode("ascii")
    fields = parse_fields(text, command=self.name, expected_count=2)
    unit_id, flag_s = fields
    return BlinkDisplayState(
        unit_id=unit_id,
        flashing=parse_bool_code(flag_s, field=f"{self.name}.flashing"),
    )

def encode(self, ctx: DecodeContext, request: BlinkDisplayRequest) -> bytes:
    """Emit FFP query or set bytes."""
    prefix = ctx.command_prefix.decode("ascii")
    if request.duration_s is None:
        return f"{ctx.unit_id}{prefix}{self.token}\r".encode("ascii")
    if request.duration_s < 0 and request.duration_s != _FFP_FLASH_INDEFINITELY:
        raise AlicatValidationError(
            f"{self.name}: duration_s must be >= 0 or exactly "
            f"{_FFP_FLASH_INDEFINITELY} (flash indefinitely), "
            f"got {request.duration_s}",
            context=ErrorContext(
                command_name=self.name,
                unit_id=ctx.unit_id,
                extra={"duration_s": request.duration_s},
            ),
        )
    return f"{ctx.unit_id}{prefix}{self.token} {request.duration_s}\r".encode("ascii")

def decode(
    self,
    response: bytes | tuple[bytes, ...],
    ctx: DecodeContext,
) -> ParsedFrame:
    """Parse the post-op data frame."""
    return _decode_hold_frame(self.name, response, ctx)

def encode(self, ctx: DecodeContext, request: CancelValveHoldRequest) -> bytes:
    r"""Emit ``<uid><prefix>C\r``."""
    del request
    prefix = ctx.command_prefix.decode("ascii")
    return f"{ctx.unit_id}{prefix}{self.token}\r".encode("ascii")

def decode(
    self,
    response: bytes | tuple[bytes, ...],
    ctx: DecodeContext,
) -> Resp:
    """Parse the device's (EOL-stripped) response into the typed result."""
    raise NotImplementedError(f"{self.name}.decode is not implemented")

def encode(self, ctx: DecodeContext, request: Req) -> bytes:
    """Render ``request`` as the exact bytes to put on the wire (incl. EOL)."""
    raise NotImplementedError(f"{self.name}.encode is not implemented")

def decode(
    self,
    response: bytes | tuple[bytes, ...],
    ctx: DecodeContext,
) -> DataFrameFormat:
    """Parse the advertised data-frame layout into :class:`DataFrameFormat`."""
    del ctx
    if not isinstance(response, tuple):
        raise TypeError(
            f"{self.name}.decode expected multi-line response, got single line",
        )
    return parse_data_frame_table(response)

def encode(
    self,
    ctx: DecodeContext,
    request: DataFrameFormatRequest,
) -> bytes:
    r"""Emit ``<unit_id><prefix>??D*\r``."""
    del request
    prefix = ctx.command_prefix.decode("ascii")
    return f"{ctx.unit_id}{prefix}{self.token}\r".encode("ascii")

def decode(
    self,
    response: bytes | tuple[bytes, ...],
    ctx: DecodeContext,
) -> DeadbandSetting:
    """Parse the 4-field reply into :class:`DeadbandSetting`."""
    del ctx
    if isinstance(response, tuple):
        raise TypeError(
            f"{self.name}.decode expected single-line response, got {len(response)} lines",
        )
    text = response.decode("ascii")
    fields = parse_fields(text, command=self.name, expected_count=4)
    unit_id, deadband_s, unit_code_s, unit_label = fields
    deadband = parse_float(deadband_s, field=f"{self.name}.deadband")
    unit_code = parse_int(unit_code_s, field=f"{self.name}.unit_code")
    return DeadbandSetting(
        unit_id=unit_id,
        deadband=deadband,
        unit_code=unit_code,
        unit=_resolve_unit_label(unit_label, unit_code),
        unit_label=unit_label,
    )

def encode(self, ctx: DecodeContext, request: DeadbandLimitRequest) -> bytes:
    """Emit the LCDB query or set bytes."""
    prefix = ctx.command_prefix.decode("ascii")
    if request.deadband is None:
        # Query — ``save`` is set-only semantics, silently ignored
        # here so a caller that reuses a request dataclass with
        # default flags for both query and set doesn't have to
        # reset them.
        return f"{ctx.unit_id}{prefix}{self.token}\r".encode("ascii")
    if request.deadband < 0:
        raise AlicatValidationError(
            f"{self.name}: deadband must be >= 0 (0 disables), got {request.deadband}",
            context=ErrorContext(
                command_name=self.name,
                unit_id=ctx.unit_id,
                extra={"deadband": request.deadband},
            ),
        )
    save_flag = "1" if request.save else "0"
    return (f"{ctx.unit_id}{prefix}{self.token} {save_flag} {request.deadband}\r").encode(
        "ascii"
    )

def decode(
    self,
    response: bytes | tuple[bytes, ...],
    ctx: DecodeContext,
) -> UnitSetting:
    """Parse ``<uid> <unit_code> <unit_label>`` into :class:`UnitSetting`.

    Verified against a V10 capture on 2026-04-17 (design §16.6) — the
    device does *not* echo the requested statistic in the reply, so
    ``statistic`` is left as :attr:`Statistic.NONE` and the facade
    fills it from the request via :func:`dataclasses.replace`.
    """
    del ctx
    if isinstance(response, tuple):
        raise TypeError(
            f"{self.name}.decode expected single-line response, got {len(response)} lines",
        )
    text = response.decode("ascii")
    fields = parse_fields(text, command=self.name, expected_count=3)
    unit_id, unit_code_s, label = fields
    unit_code = parse_int(unit_code_s, field="unit_code")

    return UnitSetting(
        unit_id=unit_id,
        statistic=Statistic.NONE,  # facade replaces with request.statistic
        unit=_resolve_response_unit(label, unit_code),
        label=label,
    )

def encode(
    self,
    ctx: DecodeContext,
    request: EngineeringUnitsRequest,
) -> bytes:
    """Emit the DCU query / set bytes."""
    _, stat_code = _resolve_statistic_code(request.statistic)
    prefix = ctx.command_prefix.decode("ascii")
    head = f"{ctx.unit_id}{prefix}{self.token} {stat_code}"

    if request.unit is None:
        # Query form. ``apply_to_group`` / ``override_special_rules``
        # are set-only semantics; silently ignored here so a caller
        # that constructs the request with default flags for both
        # query and set doesn't have to reset them.
        return (head + "\r").encode("ascii")

    unit_code = _resolve_unit_code(request.unit)
    tokens = [head, str(unit_code)]
    if request.apply_to_group or request.override_special_rules:
        tokens.append("1" if request.apply_to_group else "0")
        if request.override_special_rules:
            tokens.append("1")
    return (" ".join(tokens) + "\r").encode("ascii")

def decode(
    self,
    response: bytes | tuple[bytes, ...],
    ctx: DecodeContext,
) -> FullScaleValue:
    """Parse ``<uid> <value> <unit_code> <unit_label>`` into :class:`FullScaleValue`.

    Verified against a V10 capture on 2026-04-17 (design §16.6) — the
    device does *not* echo the requested statistic in the reply, so
    ``statistic`` is left as :attr:`Statistic.NONE` and the facade
    fills it from the request via :func:`dataclasses.replace`.
    """
    del ctx
    if isinstance(response, tuple):
        raise TypeError(
            f"{self.name}.decode expected single-line response, got {len(response)} lines",
        )
    text = response.decode("ascii")
    fields = parse_fields(text, command=self.name, expected_count=4)
    _unit_id, value_s, unit_code_s, label = fields
    value = parse_float(value_s, field="full_scale_value")
    unit_code = parse_int(unit_code_s, field="unit_code")

    return FullScaleValue(
        statistic=Statistic.NONE,  # facade replaces with request.statistic
        value=value,
        unit=_resolve_response_unit(label, unit_code),
        unit_label=label,
    )

def encode(
    self,
    ctx: DecodeContext,
    request: FullScaleQueryRequest,
) -> bytes:
    r"""Emit ``<unit_id><prefix>FPF <stat_code>\r``."""
    _, stat_code = _resolve_statistic_code(request.statistic)
    prefix = ctx.command_prefix.decode("ascii")
    return f"{ctx.unit_id}{prefix}{self.token} {stat_code}\r".encode("ascii")

def decode(
    self,
    response: bytes | tuple[bytes, ...],
    ctx: DecodeContext,
) -> dict[int, str]:
    """Parse the multi-line gas list into ``{gas_code: label}``."""
    del ctx
    if not isinstance(response, tuple):
        raise TypeError(
            f"{self.name}.decode expected multi-line response, got single line",
        )
    return parse_gas_list(response)

def encode(self, ctx: DecodeContext, request: GasListRequest) -> bytes:
    r"""Emit ``<unit_id><prefix>??G*\r``."""
    del request
    prefix = ctx.command_prefix.decode("ascii")
    return f"{ctx.unit_id}{prefix}{self.token}\r".encode("ascii")

def decode(
    self,
    response: bytes | tuple[bytes, ...],
    ctx: DecodeContext,
) -> GasState:
    """Parse the four-field GS reply into a typed :class:`GasState`."""
    del ctx  # kept for signature uniformity across commands
    if isinstance(response, tuple):
        # A LINE command should never receive a multi-line response; the
        # session guarantees this, but guard against a mis-configured spec.
        raise TypeError(
            f"{self.name}.decode expected single-line response, got {len(response)} lines",
        )
    text = response.decode("ascii")
    fields = parse_fields(text, command=self.name, expected_count=4)
    unit_id, code_s, label, long_name = fields
    code = parse_int(code_s, field="code")
    return GasState(
        unit_id=unit_id,
        code=code,
        gas=gas_registry.by_code(code),
        label=label,
        long_name=long_name,
    )

def encode(self, ctx: DecodeContext, request: GasSelectRequest) -> bytes:
    """Emit the wire bytes for a GS query or set command."""
    prefix = ctx.command_prefix.decode("ascii")
    if request.gas is None:
        # Query form: no gas argument, no save flag.
        return f"{ctx.unit_id}{prefix}{self.token}\r".encode("ascii")
    gas = gas_registry.coerce(request.gas)
    if request.save is None:
        body = f"{ctx.unit_id}{prefix}{self.token} {gas.code}"
    else:
        save_flag = "1" if request.save else "0"
        body = f"{ctx.unit_id}{prefix}{self.token} {gas.code} {save_flag}"
    return body.encode("ascii") + b"\r"

def decode(
    self,
    response: bytes | tuple[bytes, ...],
    ctx: DecodeContext,
) -> ParsedFrame:
    """Parse the post-set data frame against ``ctx.data_frame_format``.

    The session caches the format at startup; callers hitting this
    decoder before the format has been probed get an
    :class:`AlicatParseError` pointing at the missing probe — same
    failure mode as :data:`~alicatlib.commands.polling.POLL_DATA`.
    """
    if isinstance(response, tuple):
        raise TypeError(
            f"{self.name}.decode expected single-line response, got {len(response)} lines",
        )
    if ctx.data_frame_format is None:
        raise AlicatParseError(
            f"{self.name} requires ctx.data_frame_format; session must probe ??D* first",
            field_name="data_frame_format",
            expected="DataFrameFormat",
            actual=None,
            context=ErrorContext(command_name=self.name, raw_response=response),
        )
    return ctx.data_frame_format.parse(response)

def encode(self, ctx: DecodeContext, request: GasSelectLegacyRequest) -> bytes:
    r"""Emit ``<unit_id><prefix>G <gas_code>\r``."""
    gas = gas_registry.coerce(request.gas)
    prefix = ctx.command_prefix.decode("ascii")
    body = f"{ctx.unit_id}{prefix}{self.token} {gas.code}"
    return body.encode("ascii") + b"\r"