jmri_loconet.codec

Frame-level codec: read whole LocoNet frames from a byte stream, verify checksums, and build new frames with correct checksums.

This module is protocol-shape only — it doesn't know what opcodes mean. The jmri_loconet.messages module sits on top and turns frame bytes into typed message dataclasses.

View Source

 1"""Frame-level codec: read whole LocoNet frames from a byte stream, verify
 2checksums, and build new frames with correct checksums.
 3
 4This module is *protocol-shape only* — it doesn't know what opcodes mean.
 5The :mod:`jmri_loconet.messages` module sits on top and turns frame bytes
 6into typed message dataclasses.
 7"""
 8from __future__ import annotations
 9
10from collections.abc import Awaitable, Callable
11
12from jmri_loconet.opcodes import (
13    expected_length,
14    is_valid_opcode_byte,
15    length_class_of,
16)
17
18
19def compute_checksum(body: bytes) -> int:
20    """Return the LocoNet checksum byte for *body* (everything before the CS).
21
22    The checksum is ``0xFF XOR`` of every byte in *body*.
23    """
24    cs = 0xFF
25    for b in body:
26        cs ^= b
27    return cs
28
29
30def verify_checksum(frame: bytes) -> bool:
31    """True if the trailing byte of *frame* is its correct checksum."""
32    if len(frame) < 2:
33        return False
34    return compute_checksum(frame[:-1]) == frame[-1]
35
36
37def append_checksum(body: bytes) -> bytes:
38    """Append a correct checksum to *body* and return the complete frame."""
39    return body + bytes([compute_checksum(body)])
40
41
42def frame_length(frame: bytes) -> int | None:
43    """Return the expected total length of *frame* given its leading opcode.
44
45    For fixed-length frames returns 2/4/6 directly. For variable-length
46    frames returns the value in ``frame[1]`` if present, else ``None``.
47    Returns ``None`` for any leading byte that isn't a valid opcode.
48    """
49    if not frame or not is_valid_opcode_byte(frame[0]):
50        return None
51    fixed = expected_length(frame[0])
52    if fixed is not None:
53        return fixed
54    if len(frame) < 2:
55        return None
56    return frame[1]
57
58
59async def read_frame(
60    read_byte: Callable[[], Awaitable[int]],
61    read_exact: Callable[[int], Awaitable[bytes]],
62) -> bytes:
63    """Read one complete LocoNet frame from a byte stream.
64
65    *read_byte* returns a single byte; *read_exact(n)* returns exactly *n*
66    bytes. Non-opcode bytes seen before an opcode (e.g. mid-message noise)
67    are silently dropped — matching JMRI's resync behaviour.
68    """
69    # Resync to the next opcode byte.
70    while True:
71        b = await read_byte()
72        if is_valid_opcode_byte(b):
73            opcode = b
74            break
75
76    cls = length_class_of(opcode)
77    if cls == 0b11:
78        length_byte = await read_byte()
79        rest = await read_exact(length_byte - 2)
80        return bytes([opcode, length_byte]) + rest
81
82    total = expected_length(opcode)
83    assert total is not None  # cls != 0b11 guarantees a known fixed length
84    rest = await read_exact(total - 1)
85    return bytes([opcode]) + rest

def compute_checksum(body: bytes) -> int: View Source

20def compute_checksum(body: bytes) -> int:
21    """Return the LocoNet checksum byte for *body* (everything before the CS).
22
23    The checksum is ``0xFF XOR`` of every byte in *body*.
24    """
25    cs = 0xFF
26    for b in body:
27        cs ^= b
28    return cs

Return the LocoNet checksum byte for body (everything before the CS).

The checksum is 0xFF XOR of every byte in body.

def verify_checksum(frame: bytes) -> bool: View Source

31def verify_checksum(frame: bytes) -> bool:
32    """True if the trailing byte of *frame* is its correct checksum."""
33    if len(frame) < 2:
34        return False
35    return compute_checksum(frame[:-1]) == frame[-1]

True if the trailing byte of frame is its correct checksum.

def append_checksum(body: bytes) -> bytes: View Source

38def append_checksum(body: bytes) -> bytes:
39    """Append a correct checksum to *body* and return the complete frame."""
40    return body + bytes([compute_checksum(body)])

Append a correct checksum to body and return the complete frame.

def frame_length(frame: bytes) -> int | None: View Source

43def frame_length(frame: bytes) -> int | None:
44    """Return the expected total length of *frame* given its leading opcode.
45
46    For fixed-length frames returns 2/4/6 directly. For variable-length
47    frames returns the value in ``frame[1]`` if present, else ``None``.
48    Returns ``None`` for any leading byte that isn't a valid opcode.
49    """
50    if not frame or not is_valid_opcode_byte(frame[0]):
51        return None
52    fixed = expected_length(frame[0])
53    if fixed is not None:
54        return fixed
55    if len(frame) < 2:
56        return None
57    return frame[1]

Return the expected total length of frame given its leading opcode.

For fixed-length frames returns 2/4/6 directly. For variable-length frames returns the value in frame[1] if present, else None. Returns None for any leading byte that isn't a valid opcode.

async def read_frame( read_byte: Callable[[], Awaitable[int]], read_exact: Callable[[int], Awaitable[bytes]]) -> bytes: View Source

60async def read_frame(
61    read_byte: Callable[[], Awaitable[int]],
62    read_exact: Callable[[int], Awaitable[bytes]],
63) -> bytes:
64    """Read one complete LocoNet frame from a byte stream.
65
66    *read_byte* returns a single byte; *read_exact(n)* returns exactly *n*
67    bytes. Non-opcode bytes seen before an opcode (e.g. mid-message noise)
68    are silently dropped — matching JMRI's resync behaviour.
69    """
70    # Resync to the next opcode byte.
71    while True:
72        b = await read_byte()
73        if is_valid_opcode_byte(b):
74            opcode = b
75            break
76
77    cls = length_class_of(opcode)
78    if cls == 0b11:
79        length_byte = await read_byte()
80        rest = await read_exact(length_byte - 2)
81        return bytes([opcode, length_byte]) + rest
82
83    total = expected_length(opcode)
84    assert total is not None  # cls != 0b11 guarantees a known fixed length
85    rest = await read_exact(total - 1)
86    return bytes([opcode]) + rest

Read one complete LocoNet frame from a byte stream.

read_byte returns a single byte; read_exact(n) returns exactly n bytes. Non-opcode bytes seen before an opcode (e.g. mid-message noise) are silently dropped — matching JMRI's resync behaviour.