pamoja.serial

  1"""Idiomatic serial-framing facade.
  2
  3A serial line is a stream of bytes with no packet boundaries, so something has to
  4mark where one message ends and the next begins. SLIP and COBS are the two ways to
  5do that, and each is offered both as a one-shot call over a complete frame and as
  6a streaming decoder for the arbitrary chunks a port delivers.
  7
  8The streaming decoders are what a real read loop uses. A corrupt frame does not
  9raise, because the frames around it are still good; it is dropped and counted on
 10:attr:`SlipDecoder.discarded`.
 11"""
 12
 13from __future__ import annotations
 14
 15from typing import Protocol
 16
 17from pamoja._native import CobsDecoder as _NativeCobsDecoder
 18from pamoja._native import SlipDecoder as _NativeSlipDecoder
 19from pamoja._native import cobs_decode as _cobs_decode
 20from pamoja._native import cobs_encode as _cobs_encode
 21from pamoja._native import cobs_max_encoded_len as _cobs_max_encoded_len
 22from pamoja._native import serial_framing_bytes as _serial_framing_bytes
 23from pamoja._native import slip_decode as _slip_decode
 24from pamoja._native import slip_encode as _slip_encode
 25from pamoja._native import slip_max_encoded_len as _slip_max_encoded_len
 26
 27__all__ = [
 28    "COBS_DELIMITER",
 29    "CobsDecoder",
 30    "Framing",
 31    "SLIP_END",
 32    "SLIP_ESC",
 33    "SLIP_ESC_END",
 34    "SLIP_ESC_ESC",
 35    "SlipDecoder",
 36    "cobs",
 37    "slip",
 38]
 39
 40
 41class Framing(Protocol):
 42    """One of the two byte-stuffing framings this module offers."""
 43
 44    def encode(self, payload: bytes) -> bytes:
 45        """Frame a payload for the wire.
 46
 47        :param payload: The bytes to send.
 48        :returns: The frame, delimiter included.
 49        """
 50
 51    def decode(self, frame: bytes) -> bytes:
 52        """Read the payload back out of a complete frame.
 53
 54        :param frame: The frame as it arrived.
 55        :returns: The payload.
 56        :raises PamojaError: If the frame is corrupt.
 57        """
 58
 59    def max_encoded_len(self, payload_len: int) -> int:
 60        """Return the largest frame a payload of this length can produce.
 61
 62        :param payload_len: The payload length in bytes.
 63        :returns: The worst-case frame length.
 64        """
 65
 66
 67_END, _ESC, _ESC_END, _ESC_ESC, _DELIMITER = _serial_framing_bytes()
 68
 69#: The SLIP byte that ends a frame (RFC 1055).
 70SLIP_END = _END
 71#: The SLIP byte that escapes a reserved value inside a frame (RFC 1055).
 72SLIP_ESC = _ESC
 73#: The byte that follows an escape to stand for a literal end byte.
 74SLIP_ESC_END = _ESC_END
 75#: The byte that follows an escape to stand for a literal escape byte.
 76SLIP_ESC_ESC = _ESC_ESC
 77#: The byte that delimits a COBS frame, which never appears inside one.
 78COBS_DELIMITER = _DELIMITER
 79
 80
 81class _Slip:
 82    """SLIP (RFC 1055): an ``END`` byte ends a packet, and an escape pair carries it."""
 83
 84    __slots__ = ()
 85
 86    def encode(self, payload: bytes) -> bytes:
 87        """Frame a payload as a SLIP packet."""
 88        return _slip_encode(bytes(payload))
 89
 90    def decode(self, frame: bytes) -> bytes:
 91        """Read the payload back out of a SLIP frame."""
 92        return _slip_decode(bytes(frame))
 93
 94    def max_encoded_len(self, payload_len: int) -> int:
 95        """Return the worst-case SLIP frame length for a payload."""
 96        return _slip_max_encoded_len(payload_len)
 97
 98
 99class _Cobs:
100    """COBS: removes the zero byte so one zero delimits packets unambiguously."""
101
102    __slots__ = ()
103
104    def encode(self, payload: bytes) -> bytes:
105        """Frame a payload as a COBS packet."""
106        return _cobs_encode(bytes(payload))
107
108    def decode(self, frame: bytes) -> bytes:
109        """Read the payload back out of a COBS frame."""
110        return _cobs_decode(bytes(frame))
111
112    def max_encoded_len(self, payload_len: int) -> int:
113        """Return the worst-case COBS frame length for a payload."""
114        return _cobs_max_encoded_len(payload_len)
115
116
117#: SLIP framing, the simplest there is.
118slip: Framing = _Slip()
119
120#: COBS framing, for links where the overhead has to stay small and predictable.
121cobs: Framing = _Cobs()
122
123
124class SlipDecoder:
125    """Reassembles whole SLIP frames from the chunks a serial port delivers.
126
127    Example::
128
129        decoder = SlipDecoder()
130        while True:
131            for frame in decoder.feed(port.read(256)):
132                handle(frame)
133    """
134
135    __slots__ = ("_native",)
136
137    def __init__(self) -> None:
138        """Create an empty decoder, ready for the first chunk."""
139        self._native = _NativeSlipDecoder()
140
141    def feed(self, chunk: bytes) -> list[bytes]:
142        """Feed a chunk of the stream.
143
144        :param chunk: The bytes just read from the port.
145        :returns: Every frame this chunk completed, in order, which is often none.
146        """
147        return self._native.feed(bytes(chunk))
148
149    @property
150    def discarded(self) -> int:
151        """How many corrupt frames this decoder has discarded."""
152        return self._native.discarded
153
154    def reset(self) -> None:
155        """Discard any partly assembled frame."""
156        self._native.reset()
157
158
159class CobsDecoder:
160    """Reassembles whole COBS frames from the chunks a serial port delivers.
161
162    The counterpart to :class:`SlipDecoder`, for links where the framing overhead
163    has to stay small and predictable.
164    """
165
166    __slots__ = ("_native",)
167
168    def __init__(self) -> None:
169        """Create an empty decoder, ready for the first chunk."""
170        self._native = _NativeCobsDecoder()
171
172    def feed(self, chunk: bytes) -> list[bytes]:
173        """Feed a chunk of the stream.
174
175        :param chunk: The bytes just read from the port.
176        :returns: Every frame this chunk completed, in order.
177        """
178        return self._native.feed(bytes(chunk))
179
180    @property
181    def discarded(self) -> int:
182        """How many corrupt frames this decoder has discarded."""
183        return self._native.discarded
184
185    def reset(self) -> None:
186        """Discard any partly assembled frame."""
187        self._native.reset()