watlowlib.protocol¶

def make_protocol_client(
    kind: ProtocolKind,
    transport: Transport,
) -> ProtocolClient[Any, Any]:
    """Build an address-agnostic :class:`ProtocolClient` for ``kind`` over ``transport``.

    The returned client takes a destination address per
    :meth:`ProtocolClient.execute` call, so one client can serve every
    device on a multi-drop RS-485 segment.

    Args:
        kind: The wire protocol. ``AUTO`` is rejected here — the
            detector must resolve it to a concrete kind first.
        transport: An open or openable :class:`Transport`. Lifecycle is
            the caller's responsibility — the client does not call
            ``open()`` on construction. For ``MODBUS_RTU`` this must be
            a :class:`ModbusBusTransport`.

    Raises:
        WatlowConfigurationError: ``kind`` is ``AUTO`` (use the
            detector), or the ``transport`` shape doesn't match
            ``kind``.
    """
    if kind is ProtocolKind.STDBUS:
        # Imported lazily so the modbus / detect branches don't pull
        # the stdbus subpackage when they aren't needed.
        from watlowlib.protocol.stdbus.client import StdBusProtocolClient  # noqa: PLC0415

        return StdBusProtocolClient(transport)
    if kind is ProtocolKind.MODBUS_RTU:
        from watlowlib.protocol.modbus.client import ModbusProtocolClient  # noqa: PLC0415
        from watlowlib.protocol.modbus.transport import ModbusBusTransport  # noqa: PLC0415

        if not isinstance(transport, ModbusBusTransport):
            raise WatlowConfigurationError(
                "ProtocolKind.MODBUS_RTU requires a ModbusBusTransport "
                f"(got {type(transport).__name__}).",
            )
        bus_transport = transport
        return ModbusProtocolClient(
            slave_provider=bus_transport.bus.slave,
            port=bus_transport.label,
        )
    if kind is ProtocolKind.AUTO:
        raise WatlowConfigurationError(
            "ProtocolKind.AUTO must be resolved by the detector before reaching "
            "make_protocol_client.",
        )
    # StrEnum is closed; an unknown value would have failed at parse time.
    raise WatlowConfigurationError(f"unsupported protocol kind: {kind!r}")

def addr_to_mac(addr: int) -> int:
    """Map a Standard Bus address (``1..16``) to its MS/TP MAC.

    Raises:
        ValueError: ``addr`` is outside ``1..16``.
    """
    if not 1 <= addr <= 16:
        raise ValueError(f"Standard Bus address out of range 1..16: {addr}")
    return ADDR_OFFSET + addr

def data_crc16(data: bytes) -> int:
    """Compute the BACnet MS/TP data CRC-16.

    Returns the host-order int; on the wire, send little-endian via
    :func:`data_crc16_le_bytes`.
    """
    crc = 0xFFFF
    for b in data:
        low = (crc & 0xFF) ^ b
        crc = (
            (crc >> 8)
            ^ (low << 8)
            ^ (low << 3)
            ^ (low << 12)
            ^ (low >> 4)
            ^ (low & 0x0F)
            ^ ((low & 0x0F) << 7)
        ) & 0xFFFF
    return (~crc) & 0xFFFF

def data_crc16_le_bytes(data: bytes) -> bytes:
    """Encode the data CRC as the two on-wire bytes (little-endian)."""
    return data_crc16(data).to_bytes(2, "little")

def decode_frame(buf: bytes) -> Frame:
    """Parse wire bytes into a :class:`Frame`, verifying both CRCs.

    The caller is expected to have already framed on the ``55 FF``
    preamble — :class:`watlowlib.protocol.stdbus.client.StdBusProtocolClient`
    handles that during read.

    Raises:
        FrameError: short buffer, wrong preamble, header CRC mismatch,
            truncated body, or data CRC mismatch.
    """
    if len(buf) < 8:
        raise FrameError(f"frame too short: {len(buf)} bytes")
    if buf[:2] != PREAMBLE:
        raise FrameError(f"bad preamble: {buf[:2].hex()}")
    frame_type = buf[2]
    dst = buf[3]
    src = buf[4]
    plen = (buf[5] << 8) | buf[6]
    hcrc = buf[7]
    expected_hcrc = header_crc8(buf[2:7])
    if hcrc != expected_hcrc:
        raise FrameError(f"header CRC mismatch: got 0x{hcrc:02X}, want 0x{expected_hcrc:02X}")
    if plen == 0:
        return Frame(frame_type, dst, src, b"")
    expected_total = 8 + plen + 2
    if len(buf) < expected_total:
        raise FrameError(f"frame truncated: have {len(buf)}, need {expected_total}")
    payload = buf[8 : 8 + plen]
    dcrc = buf[8 + plen : 8 + plen + 2]
    expected_dcrc = data_crc16_le_bytes(payload)
    if dcrc != expected_dcrc:
        raise FrameError(f"data CRC mismatch: got {dcrc.hex()}, want {expected_dcrc.hex()}")
    return Frame(frame_type, dst, src, payload)

def decode_payload(
    payload: bytes,
) -> ReadResponse | WriteResponse | ReadRequest | WriteRequest | ErrorResponse:
    """Parse an inner Watlow payload into a structured response/request/error."""
    if len(payload) < 2:
        raise ValueError(f"payload too short: {payload.hex()}")
    direction, function = payload[0], payload[1]
    if direction not in (DIR_REQUEST, DIR_RESPONSE):
        raise ValueError(f"unknown direction byte 0x{direction:02X}")

    # Error response: 0x02 followed by an error code with high bit set.
    if direction == DIR_RESPONSE and function & 0x80:
        if len(payload) != 2:
            raise ValueError(f"error response with unexpected trailing bytes: {payload.hex()}")
        return ErrorResponse(function)

    if function == FN_READ:
        if direction == DIR_REQUEST:
            if len(payload) < 6 or payload[2] != 0x01:
                raise ValueError(f"unexpected read-request shape: {payload.hex()}")
            return ReadRequest(payload[3], payload[4], payload[5])
        # read response: 02 03 01 CC MM II <typed value>
        if len(payload) < 7 or payload[2] != 0x01:
            raise ValueError(f"unexpected read-response shape: {payload.hex()}")
        cls, mem, inst = payload[3], payload[4], payload[5]
        value, tag, _ = decode_value(payload[6:])
        return ReadResponse(cls, mem, inst, value, tag)

    if function == FN_WRITE:
        # write request/response: dd 04 CC MM II <typed value>
        if len(payload) < 6:
            raise ValueError(f"unexpected write shape: {payload.hex()}")
        cls, mem, inst = payload[2], payload[3], payload[4]
        value, tag, _ = decode_value(payload[5:])
        return (
            WriteRequest(cls, mem, inst, value, tag)
            if direction == DIR_REQUEST
            else WriteResponse(cls, mem, inst, value, tag)
        )

    raise ValueError(f"unknown function byte 0x{function:02X}")

def decode_value(buf: bytes) -> tuple[float | int | str, int, int]:
    """Decode a single value starting at ``buf[0]``.

    Returns ``(value, type_tag, consumed_bytes)``.
    """
    if not buf:
        raise ValueError("empty value buffer")
    tag = buf[0]
    if tag == DataType.U8:
        if len(buf) < 2:
            raise ValueError("truncated u8 value")
        return buf[1], tag, 2
    if tag == DataType.U16:
        if len(buf) < 3:
            raise ValueError("truncated u16 value")
        (v_u16,) = struct.unpack(">H", buf[1:3])
        return v_u16, tag, 3
    if tag == DataType.U32:
        if len(buf) < 5:
            raise ValueError("truncated u32 value")
        (v_u32,) = struct.unpack(">I", buf[1:5])
        return v_u32, tag, 5
    if tag == DataType.S32:
        if len(buf) < 5:
            raise ValueError("truncated s32 value")
        (v_s32,) = struct.unpack(">i", buf[1:5])
        return v_s32, tag, 5
    if tag == DataType.FLOAT:
        if len(buf) < 5:
            raise ValueError("truncated float value")
        (v_f,) = struct.unpack(">f", buf[1:5])
        return v_f, tag, 5
    if tag == DataType.STRING:
        if len(buf) < 2:
            raise ValueError("truncated string header")
        n = buf[1]
        if len(buf) < 2 + n:
            raise ValueError("truncated string body")
        raw = bytes(buf[2 : 2 + n])
        # Trim trailing NULs for convenience but keep the raw length
        # intact in caller paths.
        s = raw.rstrip(b"\x00").decode("ascii", errors="replace")
        return s, tag, 2 + n
    if tag == DataType.PACKED:
        if len(buf) < 2:
            raise ValueError("truncated packed-int header")
        n = buf[1]  # count of 16-bit words
        if len(buf) < 2 + 2 * n:
            raise ValueError("truncated packed-int body")
        # PACKED with count=1 is the canonical enum / 16-bit case
        # (the most common shape on the wire).
        if n == 1:
            (v_packed,) = struct.unpack(">H", buf[2:4])
            return v_packed, tag, 4
        # count>=2: pack words MSW-first into a single integer
        # (provisional; only confirmed once we observe a real Wide
        # Enum that exceeds 16 bits).
        v_wide = 0
        for i in range(n):
            (w,) = struct.unpack(">H", buf[2 + 2 * i : 4 + 2 * i])
            v_wide = (v_wide << 16) | w
        return v_wide, tag, 2 + 2 * n
    raise ValueError(f"unknown type tag: 0x{tag:02X} (rest: {buf[:8].hex()})")

def encode_frame(frame: Frame) -> bytes:
    """Serialise ``frame`` to wire bytes (preamble through data CRC).

    Raises:
        FrameError: ``frame.payload`` exceeds the 16-bit length field.
    """
    if len(frame.payload) > 0xFFFF:
        raise FrameError(f"payload too long: {len(frame.payload)}")
    header = bytes(
        [
            frame.frame_type & 0xFF,
            frame.dst & 0xFF,
            frame.src & 0xFF,
            (len(frame.payload) >> 8) & 0xFF,
            len(frame.payload) & 0xFF,
        ]
    )
    hcrc = bytes([header_crc8(header)])
    if frame.payload:
        return PREAMBLE + header + hcrc + frame.payload + data_crc16_le_bytes(frame.payload)
    return PREAMBLE + header + hcrc

def encode_read_request(param_id: int, instance: int = 1) -> bytes:
    """Build the inner payload for reading a single parameter."""
    cls, member = split_param(param_id)
    return bytes([DIR_REQUEST, FN_READ, 0x01, cls, member, instance])

def encode_value(type_tag: int, value: float | int | str | bytes) -> bytes:
    """Encode ``value`` under ``type_tag`` to wire bytes (tag included)."""
    if type_tag == DataType.STRING:
        # STRING is the only branch that accepts a str/bytes payload;
        # the numeric branches all coerce via int()/float() and would
        # silently produce zero-filled buffers if a str slipped through.
        if not isinstance(value, str | bytes):
            raise TypeError(f"STRING type tag requires str or bytes, got {type(value).__name__}")
        b = value.encode("ascii") if isinstance(value, str) else bytes(value)
        if len(b) > 0xFF:
            raise ValueError(f"string too long: {len(b)}")
        return bytes([DataType.STRING, len(b)]) + b
    # Numeric branches: reject str/bytes early with a clear message.
    if isinstance(value, str | bytes):
        raise TypeError(
            f"type tag 0x{type_tag:02X} requires a numeric value, got {type(value).__name__}"
        )
    if type_tag == DataType.FLOAT:
        return bytes([DataType.FLOAT]) + struct.pack(">f", float(value))
    if type_tag == DataType.U8:
        if not 0 <= int(value) <= 0xFF:
            raise ValueError(f"u8 value out of range: {value}")
        return bytes([DataType.U8, int(value)])
    if type_tag == DataType.U16:
        if not 0 <= int(value) <= 0xFFFF:
            raise ValueError(f"u16 value out of range: {value}")
        return bytes([DataType.U16]) + struct.pack(">H", int(value))
    if type_tag == DataType.U32:
        if not 0 <= int(value) <= 0xFFFFFFFF:
            raise ValueError(f"u32 value out of range: {value}")
        return bytes([DataType.U32]) + struct.pack(">I", int(value))
    if type_tag == DataType.S32:
        if not -(2**31) <= int(value) <= 2**31 - 1:
            raise ValueError(f"s32 value out of range: {value}")
        return bytes([DataType.S32]) + struct.pack(">i", int(value))
    if type_tag == DataType.PACKED:
        if not 0 <= int(value) <= 0xFFFF:
            raise ValueError(f"packed-int(1) value out of range: {value}")
        return bytes([DataType.PACKED, 0x01]) + struct.pack(">H", int(value))
    raise ValueError(f"unsupported type tag for encode: 0x{type_tag:02X}")

def encode_write_request(
    param_id: int,
    value: float | int | str | bytes,
    *,
    instance: int = 1,
    type_tag: int = DataType.FLOAT,
) -> bytes:
    """Build the inner payload for writing a single parameter."""
    cls, member = split_param(param_id)
    head = bytes([DIR_REQUEST, FN_WRITE, cls, member, instance])
    return head + encode_value(type_tag, value)

def header_crc8(data: bytes) -> int:
    """Compute the BACnet MS/TP header CRC-8.

    Returns the on-wire byte (one's complement of the running CRC).
    """
    crc = 0xFF
    for b in data:
        x = crc ^ b
        x = x ^ (x << 1) ^ (x << 2) ^ (x << 3) ^ (x << 4) ^ (x << 5) ^ (x << 6) ^ (x << 7)
        crc = ((x & 0xFE) ^ ((x >> 8) & 0x01)) & 0xFF
    return (~crc) & 0xFF

def join_param(cls: int, member: int) -> int:
    """Recompose a Parameter ID from its ``class`` and ``member`` parts."""
    return cls * 1000 + member

def mac_to_addr(mac: int) -> int:
    """Map an MS/TP MAC (``0x10..0x1F``) back to its Standard Bus address.

    Raises:
        ValueError: ``mac`` is outside the controller range.
    """
    if not 0x10 <= mac <= 0x1F:
        raise ValueError(f"MAC out of expected controller range: 0x{mac:02X}")
    return mac - ADDR_OFFSET

def raise_for_error_code(
    code: int,
    *,
    context: ErrorContext | None = None,
) -> None:
    """Raise the typed :class:`watlowlib.errors.WatlowError` for ``code``.

    Maps Std Bus error bytes to the typed subclasses the session uses
    for :class:`Availability` updates. Unknown error codes raise
    :class:`watlowlib.errors.WatlowProtocolError`.
    """
    if code == ErrorCode.NO_SUCH_OBJECT:
        raise WatlowNoSuchObjectError(
            f"Standard Bus error 0x{code:02X} (NO_SUCH_OBJECT)",
            context=context,
        )
    if code == ErrorCode.NO_SUCH_ATTRIBUTE:
        raise WatlowNoSuchAttributeError(
            f"Standard Bus error 0x{code:02X} (NO_SUCH_ATTRIBUTE)",
            context=context,
        )
    if code == ErrorCode.NO_SUCH_INSTANCE:
        raise WatlowNoSuchInstanceError(
            f"Standard Bus error 0x{code:02X} (NO_SUCH_INSTANCE)",
            context=context,
        )
    raise WatlowProtocolError(
        f"Standard Bus error 0x{code:02X} (unknown)",
        context=context,
    )

def split_param(param_id: int) -> tuple[int, int]:
    """Split a Watlow Parameter ID into ``(class, member)``.

    The published "Parameter ID" in user manuals is
    ``Class * 1000 + Member`` — ``4001`` decodes to ``(4, 1)``,
    ``8003`` decodes to ``(8, 3)``.
    """
    return divmod(param_id, 1000)

def encoding_for(
    data_type: DataType,
    *,
    word_order_override: str | None = None,
    register_count_override: int | None = None,
) -> ModbusEncoding:
    """Return the Modbus encoding for ``data_type``.

    Args:
        data_type: The wire data-type tag from the registry.
        word_order_override: Per-row override (from
            :attr:`ParameterSpec.word_order`). ``None`` → use the
            table default.
        register_count_override: Per-row override (mainly for
            :attr:`DataType.STRING`, which is length-driven).

    Raises:
        WatlowProtocolUnsupportedError: ``data_type`` has no Modbus
            mapping yet (e.g. a future tag added to the codec but
            unmapped here).
    """
    base = _DEFAULTS.get(data_type)
    if base is None:
        msg = f"data type {data_type.name} has no Modbus encoding"
        raise WatlowProtocolUnsupportedError(msg)
    word_order = _coerce_word_order(word_order_override) or base.word_order
    register_count = register_count_override or base.register_count
    if word_order is base.word_order and register_count == base.register_count:
        return base
    return ModbusEncoding(
        register_count=register_count,
        word_order=word_order,
        byte_order=base.byte_order,
        read_fn=base.read_fn,
    )