Improved space packet parser API

Author: robamu

Diff for: CHANGELOG.md

@@ -8,6 +8,15 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 # [unreleased]
 
+# [v0.28.0] 2025-02-03
+
+## Changed
+
+- Improved the space packet parser API: The `parse_space_packets` function can simply expects
+  a byte buffer and returns results and useful context information inside a `ParseResult`
+  structure. The former `parse_space_packets` was renamed to `parse_space_packets_from_deque`
+  and now treats the provided `deque` read-only.
+
 # [v0.27.0] 2025-01-15
 
 ## Changed

Diff for: src/spacepackets/ccsds/__init__.py

@@ -6,11 +6,14 @@
     PacketId,
     PacketSeqCtrl,
     PacketType,
+    ParserResult,
     SequenceFlags,
     SpacePacket,
     SpacePacketHeader,
     SpHeader,
     get_total_space_packet_len_from_len_field,
+    parse_space_packets,
+    parse_space_packets_from_deque,
 )
 from .time import *  # noqa: F403  # re-export
 
@@ -20,9 +23,12 @@
     "PacketId",
     "PacketSeqCtrl",
     "PacketType",
+    "ParserResult",
     "SequenceFlags",
     "SpHeader",
     "SpacePacket",
     "SpacePacketHeader",
     "get_total_space_packet_len_from_len_field",
+    "parse_space_packets",
+    "parse_space_packets_from_deque",
 ]

Diff for: src/spacepackets/ccsds/spacepacket.py

@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import dataclasses
 import enum
 import struct
 from abc import ABC, abstractmethod
@@ -553,68 +554,100 @@ def get_total_space_packet_len_from_len_field(len_field: int) -> int:
     return len_field + SPACE_PACKET_HEADER_SIZE + 1
 
 
-def parse_space_packets(
-    analysis_queue: deque[bytearray], packet_ids: Sequence[PacketId]
-) -> list[bytearray]:
-    """Given a deque of bytearrays, parse for space packets. This funtion expects the deque
-    to be filled on the right side, for example with :py:meth:`collections.deque.append`.
-    If a split packet with a valid header is detected, this function will re-insert the header into
-    the given deque on the right side.
+@dataclasses.dataclass
+class ParserResult:
+    #: List of parsed space packets.
+    tm_list: list[bytes]
+    #: Range of bytes which were skipped during parsing. This can happen if there are
+    #: broken/invalid spacepackets or packets/bytes which are not CCSDS spacepackets in the
+    #: datastream. This context information can be used for debugging.
+    skipped_ranges: list[range]
+    #: Number of bytes scanned. Incomplete packets will not increment this number. Therefore,
+    #: this can be smaller than the total size of the provided buffer. Furthermore, the packet
+    #: parser will not scan fragments at the buffer end which are smaller than the minimum CCSDS
+    #: space packet size.
+    scanned_bytes: int
 
-    :param analysis_queue:
-    :param packet_ids:
-    :return:
+    @classmethod
+    def empty(cls) -> ParserResult:
+        return ParserResult(tm_list=[], skipped_ranges=[], scanned_bytes=0)
+
+    def num_of_found_packets(self) -> int:
+        return len(self.tm_list)
+
+
+def parse_space_packets_from_deque(
+    analysis_queue: deque[bytearray | bytes], packet_ids: Sequence[PacketId]
+) -> ParserResult:
+    """Given a deque of bytearrays, parse for space packets which start with the provided
+    list of packet IDs.
+
+    This funtion expects the deque to be filled on the right side, for example with
+    :py:meth:`collections.deque.append`. This function only reads the given deque. It fills
+    the provided queue content into a regular byte buffer and then calls
+    :py:meth:`parse_space_packets`.
+
+    The user needs to take care to clear the analysis queue depending on how this API is used. The
+    number of bytes scanned is returned as part of the :py:class:`ParserResult` and can be used to
+    clear the deque or only keep relevant portions for the next parse call.
     """
-    ids_raw = [packet_id.raw() for packet_id in packet_ids]
-    tm_list = []
-    concatenated_packets = bytearray()
+    packet_buf = bytearray()
     if not analysis_queue:
-        return tm_list
-    while analysis_queue:
+        return ParserResult.empty()
+    for value in analysis_queue:
         # Put it all in one buffer
-        concatenated_packets.extend(analysis_queue.popleft())
+        packet_buf.extend(value)
+    return parse_space_packets(packet_buf, packet_ids)
+
+
+def parse_space_packets(buf: bytearray | bytes, packet_ids: Sequence[PacketId]) -> ParserResult:
+    """Given a byte buffer, parse for space packets which start with the provided list of packet
+    IDs.
+
+    If there are broken/invalid spacepackets or packets/bytes which are not CCSDS spacepackets in
+    the datastream, those bytes will be skipped,  and the range of skipped bytes is returned as
+    part of the :py:class:`ParserResult`. This context information can be used for debugging or
+    management of invalid data.
+
+    In case of incomplete packets where are partial packet header is found at the end of the buffer,
+    the scanned bytes number will be the start of the incomplete packet. The number of scanned bytes
+    returned might also be smaller than the buffer size for invalid data because the packet parser
+    will not scan fragments at the end of the buffer which are smaller than the minimum
+    space packet header length.
+    """
+    tm_list = []
+    skipped_ranges = []
+
+    packet_id_list = [packet_id.raw() for packet_id in packet_ids]
     current_idx = 0
-    if len(concatenated_packets) < 6:
-        return tm_list
-    # Packet ID detected
+    skip_start = None
+    if len(buf) < 6:
+        return ParserResult(tm_list, skipped_ranges, current_idx)
     while True:
-        # Can't even parse CCSDS header. Wait for more data to arrive.
-        if current_idx + CCSDS_HEADER_LEN >= len(concatenated_packets):
+        # Can't even parse CCSDS packet ID. Wait for more data to arrive.
+        if current_idx + CCSDS_HEADER_LEN >= len(buf):
+            if skip_start is not None:
+                skipped_ranges.append(range(skip_start, current_idx))
             break
         current_packet_id = (
-            struct.unpack("!H", concatenated_packets[current_idx : current_idx + 2])[0]
-            & PACKET_ID_MASK
+            struct.unpack("!H", buf[current_idx : current_idx + 2])[0] & PACKET_ID_MASK
         )
-        if current_packet_id in ids_raw:
-            result, current_idx = __handle_packet_id_match(
-                concatenated_packets=concatenated_packets,
-                analysis_queue=analysis_queue,
-                current_idx=current_idx,
-                tm_list=tm_list,
+        # Packet ID detected
+        if current_packet_id in packet_id_list:
+            total_packet_len = get_total_space_packet_len_from_len_field(
+                struct.unpack("!H", buf[current_idx + 4 : current_idx + 6])[0]
             )
-            if result != 0:
+            # Partial header.
+            if current_idx + total_packet_len > len(buf):
                 break
+            tm_list.append(buf[current_idx : current_idx + total_packet_len])
+            if skip_start is not None:
+                skipped_ranges.append(range(skip_start, current_idx))
+                skip_start = None
+            current_idx += total_packet_len
         else:
+            if skip_start is None:
+                skip_start = current_idx
             # Keep parsing until a packet ID is found
             current_idx += 1
-    return tm_list
-
-
-def __handle_packet_id_match(
-    concatenated_packets: bytearray,
-    analysis_queue: deque[bytearray],
-    current_idx: int,
-    tm_list: list[bytearray],
-) -> tuple[int, int]:
-    total_packet_len = get_total_space_packet_len_from_len_field(
-        struct.unpack("!H", concatenated_packets[current_idx + 4 : current_idx + 6])[0]
-    )
-    # Might be part of packet. Put back into analysis queue as whole
-    if current_idx + total_packet_len > len(concatenated_packets):
-        # Clear the queue first. We are done with parsing
-        analysis_queue.clear()
-        analysis_queue.append(concatenated_packets[current_idx:])
-        return -1, current_idx
-    tm_list.append(concatenated_packets[current_idx : current_idx + total_packet_len])
-    current_idx += total_packet_len
-    return 0, current_idx
+    return ParserResult(tm_list, skipped_ranges, current_idx)

Diff for: tests/ccsds/test_sp_parser.py

@@ -2,7 +2,7 @@
 from unittest import TestCase
 
 from spacepackets.ccsds import CdsShortTimestamp
-from spacepackets.ccsds.spacepacket import parse_space_packets
+from spacepackets.ccsds.spacepacket import parse_space_packets, parse_space_packets_from_deque
 from spacepackets.ecss.tm import PusTm
 
 
@@ -18,14 +18,33 @@ def setUp(self) -> None:
         self.packet_ids = (self.tm_packet.packet_id,)
         self.tm_packet_raw = self.tm_packet.pack()
         self.packet_deque = deque()
+        self.packet_buf = bytearray()
 
     def test_sp_parser(self):
-        self.packet_deque.appendleft(self.tm_packet_raw)
-        self.packet_deque.appendleft(self.tm_packet_raw)
-        sp_list = parse_space_packets(analysis_queue=self.packet_deque, packet_ids=self.packet_ids)
-        self.assertEqual(len(sp_list), 2)
-        self.assertEqual(sp_list[0], self.tm_packet_raw)
-        self.assertEqual(sp_list[1], self.tm_packet_raw)
+        self.packet_buf.extend(self.tm_packet_raw)
+        self.packet_buf.extend(self.tm_packet_raw)
+
+        result = parse_space_packets(buf=self.packet_buf, packet_ids=self.packet_ids)
+        self.assertEqual(len(result.tm_list), 2)
+        self.assertEqual(result.tm_list[0], self.tm_packet_raw)
+        self.assertEqual(result.tm_list[1], self.tm_packet_raw)
+        self.assertEqual(result.scanned_bytes, len(self.tm_packet_raw) * 2)
+        self.assertEqual(len(result.skipped_ranges), 0)
+
+    def test_sp_parser_with_deque(self):
+        self.packet_deque.append(self.tm_packet_raw)
+        self.packet_deque.append(self.tm_packet_raw)
+        result = parse_space_packets_from_deque(self.packet_deque, self.packet_ids)
+        self.assertEqual(len(result.tm_list), 2)
+        self.assertEqual(result.tm_list[0], self.tm_packet_raw)
+        self.assertEqual(result.tm_list[1], self.tm_packet_raw)
+        self.assertEqual(result.scanned_bytes, len(self.tm_packet_raw) * 2)
+        self.assertEqual(len(result.skipped_ranges), 0)
+        self.assertEqual(len(self.packet_deque), 2)
+        flattened_deque = bytearray()
+        while self.packet_deque:
+            flattened_deque.extend(self.packet_deque.popleft())
+        self.assertEqual(len(flattened_deque), len(self.tm_packet_raw) * 2)
 
     def test_sp_parser_crap_data_is_skipped(self):
         other_larger_packet = PusTm(
@@ -36,42 +55,52 @@ def test_sp_parser_crap_data_is_skipped(self):
             timestamp=CdsShortTimestamp.empty().pack(),
         )
         other_larger_packet_raw = other_larger_packet.pack()
-        self.packet_deque.append(self.tm_packet_raw)
-        self.packet_deque.append(bytearray(8))
-        self.packet_deque.append(other_larger_packet_raw)
-        sp_list = parse_space_packets(analysis_queue=self.packet_deque, packet_ids=self.packet_ids)
-        self.assertEqual(len(sp_list), 2)
-        self.assertEqual(sp_list[0], self.tm_packet_raw)
-        self.assertEqual(sp_list[1], other_larger_packet_raw)
+        self.packet_buf.extend(self.tm_packet_raw)
+        # Crap data, could also be a CCSDS packet with an unknown packet ID.
+        self.packet_buf.extend(bytearray(8))
+        self.packet_buf.extend(other_larger_packet_raw)
+        result = parse_space_packets(self.packet_buf, packet_ids=self.packet_ids)
+        self.assertEqual(len(result.tm_list), 2)
+        self.assertEqual(result.tm_list[0], self.tm_packet_raw)
+        self.assertEqual(result.tm_list[1], other_larger_packet_raw)
+        self.assertEqual(
+            result.skipped_ranges, [range(len(self.tm_packet_raw), len(self.tm_packet_raw) + 8)]
+        )
 
     def test_sp_parser_crap_data(self):
-        self.packet_deque.appendleft(bytearray(3))
-        sp_list = parse_space_packets(analysis_queue=self.packet_deque, packet_ids=self.packet_ids)
-        self.assertEqual(len(sp_list), 0)
-        sp_list = parse_space_packets(analysis_queue=self.packet_deque, packet_ids=self.packet_ids)
-        self.assertEqual(len(sp_list), 0)
+        self.packet_buf.extend(bytearray(3))
+        result = parse_space_packets(self.packet_buf, packet_ids=self.packet_ids)
+        self.assertEqual(len(result.tm_list), 0)
+        self.assertEqual(result.scanned_bytes, 0)
+        self.assertEqual(result.skipped_ranges, [])
+
+        self.packet_buf = bytearray(7)
+        result = parse_space_packets(self.packet_buf, packet_ids=self.packet_ids)
+        self.assertEqual(len(result.tm_list), 0)
+        # Scanned one byte
+        self.assertEqual(result.scanned_bytes, 1)
+        self.assertEqual(result.skipped_ranges, [range(1)])
 
     def test_broken_packet(self):
         # slice TM packet in half
         tm_packet_first_half = self.tm_packet_raw[:10]
         tm_packet_second_half = self.tm_packet_raw[10:]
-        self.packet_deque.appendleft(tm_packet_first_half)
-        sp_list = parse_space_packets(analysis_queue=self.packet_deque, packet_ids=self.packet_ids)
-        self.assertEqual(len(sp_list), 0)
-        self.assertEqual(len(self.packet_deque), 1)
-        self.packet_deque.append(tm_packet_second_half)
-        sp_list = parse_space_packets(analysis_queue=self.packet_deque, packet_ids=self.packet_ids)
-        self.assertEqual(len(sp_list), 1)
-        self.assertEqual(len(self.packet_deque), 0)
-        self.assertEqual(sp_list[0], self.tm_packet_raw)
+        self.packet_buf.extend(tm_packet_first_half)
+        result = parse_space_packets(self.packet_buf, packet_ids=self.packet_ids)
+        self.assertEqual(len(result.tm_list), 0)
+        self.assertEqual(result.scanned_bytes, 0)
+        self.packet_buf.extend(tm_packet_second_half)
+        result = parse_space_packets(self.packet_buf, packet_ids=self.packet_ids)
+        self.assertEqual(len(result.tm_list), 1)
+        self.assertEqual(result.tm_list[0], self.tm_packet_raw)
+        self.assertEqual(result.scanned_bytes, len(self.tm_packet_raw))
 
     def test_broken_packet_at_end(self):
-        self.packet_deque.append(self.tm_packet_raw)
+        self.packet_buf.extend(self.tm_packet_raw)
         # slice TM packet in half
         tm_packet_first_half = self.tm_packet_raw[:10]
-        self.packet_deque.append(tm_packet_first_half)
-        sp_list = parse_space_packets(analysis_queue=self.packet_deque, packet_ids=self.packet_ids)
-        self.assertEqual(len(sp_list), 1)
-        self.assertEqual(len(self.packet_deque), 1)
-        self.assertEqual(self.packet_deque.pop(), tm_packet_first_half)
-        self.assertEqual(sp_list[0], self.tm_packet_raw)
+        self.packet_buf.extend(tm_packet_first_half)
+        result = parse_space_packets(self.packet_buf, packet_ids=self.packet_ids)
+        self.assertEqual(len(result.tm_list), 1)
+        self.assertEqual(result.tm_list[0], self.tm_packet_raw)
+        self.assertEqual(result.scanned_bytes, len(self.tm_packet_raw))