Add support for filelike objects in create_dataset

- Refactor generator definitions into generators subpackage
- Use singledispatch to read packet files in xarr.py
- Add singledispatch for setting up generator binary reader
- Cleanup typehinting in definitions.py
- Add tests for generators module

Author: medley56

examples/csv_to_xtce_conversion.py

@@ -17,7 +17,7 @@
 import warnings
 from pathlib import Path
 
-from space_packet_parser import ccsds
+from space_packet_parser import generators
 from space_packet_parser.xtce import containers, definitions, encodings, parameter_types, parameters
 
 # This regex is for detecting a dynamically sized field where its bit_length is
@@ -170,7 +170,7 @@ def convert_ccsdspy_to_xtce(csv_path: Path) -> definitions.XtcePacketDefinition:
 
     packet_file = jpss_test_data_dir / "J01_G011_LZ_2021-04-09T00-00-00Z_V01.DAT1"
     with packet_file.open("rb") as packet_fh:
-        ccsds_generator = ccsds.ccsds_generator(packet_fh)
+        ccsds_generator = generators.ccsds_generator(packet_fh)
         packets = [xtce_definition.parse_bytes(binary_data) for binary_data in ccsds_generator]
 
     assert len(packets) == 7200  # noqa S101

examples/parsing_and_plotting_idex_waveforms_from_socket.py

@@ -21,7 +21,7 @@
 
 import matplotlib.pyplot as plt
 
-from space_packet_parser import ccsds
+from space_packet_parser import generators
 from space_packet_parser.xtce import definitions
 
 
@@ -116,7 +116,7 @@ def plot_full_event(data: dict):
     p.start()
 
     # Create a packet generator that listens to a socket
-    idex_ccsds_generator = ccsds.ccsds_generator(receiver)
+    idex_ccsds_generator = generators.ccsds_generator(receiver)
     # No data yet. We start recording data from an event when we encounter a packet with IDX__SCI0TYPE==1
     data: dict[int, bytes] = {}
     try:

space_packet_parser/__init__.py

@@ -3,8 +3,8 @@
 from pathlib import Path
 from typing import Union
 
-from space_packet_parser.ccsds import ccsds_generator
 from space_packet_parser.common import SpacePacket
+from space_packet_parser.generators import ccsds_generator
 from space_packet_parser.xtce.definitions import XtcePacketDefinition
 from space_packet_parser.xtce.validation import validate_xtce
 
@@ -20,6 +20,8 @@
 def load_xtce(filename: Union[str, Path]) -> XtcePacketDefinition:
     """Create an XtcePacketDefinition object from an XTCE XML file
 
+    This is a shortcut for calling XtcePacketDefinition.from_xtce().
+
     Parameters
     ----------
     filename : Union[str, Path]

space_packet_parser/cli.py

@@ -23,8 +23,8 @@
 from rich.table import Table
 from rich.tree import Tree
 
-from space_packet_parser import ccsds
-from space_packet_parser.ccsds import ccsds_generator
+from space_packet_parser import generators
+from space_packet_parser.generators import ccsds_generator
 from space_packet_parser.xtce.definitions import DEFAULT_ROOT_CONTAINER, XtcePacketDefinition
 from space_packet_parser.xtce.validation import validate_xtce
 
@@ -198,7 +198,7 @@ def parse(
     logging.debug(f"Using packet definition file: {definition_file}")
     packet_definition = XtcePacketDefinition.from_xtce(definition_file)
     with open(packet_file, "rb") as f:
-        ccsds_generator = ccsds.ccsds_generator(f, skip_header_bytes=skip_header_bytes)
+        ccsds_generator = generators.ccsds_generator(f, skip_header_bytes=skip_header_bytes)
         packets = [packet_definition.parse_bytes(binary_data) for binary_data in ccsds_generator]
 
     if packet is not None:

space_packet_parser/common.py

@@ -1,15 +1,10 @@
 """Common mixins"""
 
-import datetime as dt
 import inspect
-import io
 import logging
-import socket
-import time
 import warnings
 from abc import ABCMeta, abstractmethod
-from collections.abc import Iterator
-from typing import BinaryIO, Optional, Protocol, Union
+from typing import Optional, Protocol, Union
 
 import lxml.etree as ElementTree
 from lxml.builder import ElementMaker
@@ -387,162 +382,6 @@ def _read_from_binary_as_int(self, nbits: int) -> int:
         return int_data
 
 
-def fixed_length_generator(
-    binary_data: Union[BinaryIO, socket.socket, bytes],
-    *,
-    packet_length_bytes: int,
-    buffer_read_size_bytes: Optional[int] = None,
-    show_progress: bool = False,
-) -> Iterator[bytes]:
-    """A generator that yields fixed-length chunks from binary_data.
-
-    Parameters
-    ----------
-    binary_data : Union[BinaryIO, socket.socket, bytes]
-        Binary data source.
-    packet_length_bytes : int
-        Number of bytes per packet to yield.
-    buffer_read_size_bytes : int, optional
-        Number of bytes to read from the source per read.
-    show_progress : bool
-        If True, prints a status bar.
-
-    Yields
-    ------
-    bytes
-        Fixed-length packet bytes.
-    """
-    n_bytes_parsed = 0  # Keep track of how many bytes we have parsed
-    n_packets_parsed = 0  # Keep track of how many packets we have parsed
-    read_buffer, total_length_bytes, read_bytes_from_source, buffer_read_size_bytes = _setup_binary_reader(
-        binary_data, buffer_read_size_bytes
-    )
-    current_pos = 0  # Keep track of where we are in the buffer
-    start_time = time.time_ns()
-    while True:
-        if total_length_bytes and n_bytes_parsed == total_length_bytes:
-            break
-        if show_progress:
-            _print_progress(
-                current_bytes=n_bytes_parsed,
-                total_bytes=total_length_bytes,
-                start_time_ns=start_time,
-                current_packets=n_packets_parsed,
-            )
-        if current_pos > 20_000_000:
-            # Only trim the buffer after 20 MB read to prevent modifying
-            # the bitstream and trimming after every packet
-            read_buffer = read_buffer[current_pos:]
-            current_pos = 0
-        while len(read_buffer) - current_pos < packet_length_bytes:
-            result = read_bytes_from_source(buffer_read_size_bytes)
-            if not result:
-                break
-            read_buffer += result
-        packet_bytes = read_buffer[current_pos : current_pos + packet_length_bytes]
-        current_pos += packet_length_bytes
-        n_packets_parsed += 1
-        n_bytes_parsed += packet_length_bytes
-        yield packet_bytes
-    if show_progress:
-        _print_progress(
-            current_bytes=n_bytes_parsed,
-            total_bytes=total_length_bytes,
-            start_time_ns=start_time,
-            current_packets=n_packets_parsed,
-            end="\n",
-            log=True,
-        )
-
-
-def _setup_binary_reader(binary_data, buffer_read_size_bytes=None):
-    """Helper to set up reading from binary_data (file, socket, bytes). Returns:
-    read_buffer, total_length_bytes, read_bytes_from_source, buffer_read_size_bytes
-    """
-    read_buffer = b""
-    # ========
-    # Set up the reader based on the type of binary_data
-    # ========
-    if isinstance(binary_data, io.BufferedIOBase):
-        if buffer_read_size_bytes is None:
-            # Default to a full read of the file
-            buffer_read_size_bytes = -1
-        total_length_bytes = binary_data.seek(0, io.SEEK_END)  # This is probably preferable to len
-        binary_data.seek(0, 0)
-        logger.info(
-            f"Creating packet generator from a filelike object, {binary_data}. "
-            f"Total length is {total_length_bytes} bytes"
-        )
-        read_bytes_from_source = binary_data.read
-    elif isinstance(binary_data, socket.socket):  # It's a socket and we don't know how much data we will get
-        logger.info("Creating packet generator to read from a socket. Total length to parse is unknown.")
-        total_length_bytes = None  # We don't know how long it is
-        if buffer_read_size_bytes is None:
-            # Default to 4096 bytes from a socket
-            buffer_read_size_bytes = 4096
-        read_bytes_from_source = binary_data.recv
-    elif isinstance(binary_data, bytes):
-        read_buffer = binary_data
-        total_length_bytes = len(read_buffer)
-        read_bytes_from_source = None  # No data to read, we've filled the read_buffer already
-        logger.info(f"Creating packet generator from a bytes object. Total length is {total_length_bytes} bytes")
-    elif isinstance(binary_data, io.TextIOWrapper):
-        raise OSError("Packet data file opened in TextIO mode. You must open packet data in binary mode.")
-    else:
-        raise OSError(f"Unrecognized data source: {binary_data}")
-    return read_buffer, total_length_bytes, read_bytes_from_source, buffer_read_size_bytes
-
-
-def _print_progress(
-    *,
-    current_bytes: int,
-    total_bytes: Optional[int],
-    start_time_ns: int,
-    current_packets: int,
-    end: str = "\r",
-    log: bool = False,
-):
-    """Prints a progress bar, including statistics on parsing rate.
-
-    Parameters
-    ----------
-    current_bytes : int
-        Number of bytes parsed so far.
-    total_bytes : Optional[int]
-        Number of total bytes to parse, if known. None otherwise.
-    current_packets : int
-        Number of packets parsed so far.
-    start_time_ns : int
-        Start time on system clock, in nanoseconds.
-    end : str
-        Print function end string. Default is `\\r` to create a dynamically updating loading bar.
-    log : bool
-        If True, log the progress bar at INFO level.
-    """
-    progress_char = "="
-    bar_length = 20
-
-    if total_bytes is not None:  # If we actually have an endpoint (i.e. not using a socket)
-        percentage = int((current_bytes / total_bytes) * 100)  # Percent Completed Calculation
-        progress = int((bar_length * current_bytes) / total_bytes)  # Progress Done Calculation
-    else:
-        percentage = "???"
-        progress = 0
-
-    # Fast calls initially on Windows can result in a zero elapsed time
-    elapsed_ns = max(time.time_ns() - start_time_ns, 1)
-    delta = dt.timedelta(microseconds=elapsed_ns / 1e3)
-    kbps = int(current_bytes * 8e6 / elapsed_ns)  # 8 bits per byte, 1E9 s per ns, 1E3 bits per kb
-    pps = int(current_packets * 1e9 / elapsed_ns)
-    info_str = (
-        f"[Elapsed: {delta}, Parsed {current_bytes} bytes ({current_packets} packets) at {kbps}kb/s ({pps}pkts/s)]"
-    )
-    loadbar = f"Progress: [{progress * progress_char:{bar_length}}]{percentage}% {info_str}"
-    print(loadbar, end=end)
-    if log:
-        logger.info(loadbar)
-
-
 def _extract_bits(data: bytes, start_bit: int, nbits: int):
     """Extract nbits from the data starting from the least significant end.
 

space_packet_parser/generators/__init__.py

@@ -0,0 +1,6 @@
+"""Generators subpackage, containing packet generators for different packet formats."""
+
+from space_packet_parser.generators.ccsds import ccsds_generator
+from space_packet_parser.generators.fixed_length import fixed_length_generator
+
+__all__ = ["ccsds_generator", "fixed_length_generator"]

space_packet_parser/ccsds.py renamed to space_packet_parser/generators/ccsds.py

@@ -1,4 +1,4 @@
-"""Parsing utilities for CCSDS packets.
+"""Packet generator utilities for CCSDS packets.
 
 The parsing begins with binary data representing CCSDS Packets. A user can then create a generator
 from the binary data reading from a filelike object or a socket. The ``ccsds_generator`` function yields
@@ -16,7 +16,8 @@ class can be used to inspect the CCSDS header fields of the packet, but it does
 from enum import IntEnum
 from typing import BinaryIO, Optional, Union
 
-from space_packet_parser.common import SpacePacket, _print_progress, _setup_binary_reader
+from space_packet_parser.common import SpacePacket
+from space_packet_parser.generators.utils import _print_progress, _setup_binary_reader
 
 logger = logging.getLogger(__name__)
 
@@ -31,7 +32,7 @@ class SequenceFlags(IntEnum):
 
 
 class CCSDSPacketBytes(bytes):
-    """Binary representation of a CCSDS packet.
+    """Binary (bytes) representation of a CCSDS packet.
 
     Methods to extract the header fields are added to the raw bytes object.
     """
@@ -170,6 +171,10 @@ def create_ccsds_packet(
     -------
     : CCSDSPacketBytes
         Resulting binary packet
+
+    Notes
+    -----
+    This function is extremely useful for generating test packets for debugging or mocking purposes.
     """
     if version_number < 0 or version_number > 7:  # 3 bits
         raise ValueError("version_number must be between 0 and 7")
@@ -223,7 +228,7 @@ class to extract the header fields with specific methods.
     def __init__(self, *args, **kwargs):
         warnings.warn(
             "The CCSDSPacket class is deprecated and will be removed in a future release. "
-            "Use the Packet class instead (no CCSDS prefix).",
+            "Use the SpacePacket class instead (no CCSDS prefix).",
             DeprecationWarning,
             stacklevel=2,
         )

space_packet_parser/generators/fixed_length.py

@@ -0,0 +1,76 @@
+"""Fixed length packet generator."""
+
+import socket
+import time
+from collections.abc import Iterator
+from typing import BinaryIO, Optional, Union
+
+from space_packet_parser.generators.utils import _print_progress, _setup_binary_reader
+
+
+def fixed_length_generator(
+    binary_data: Union[BinaryIO, socket.socket, bytes],
+    *,
+    packet_length_bytes: int,
+    buffer_read_size_bytes: Optional[int] = None,
+    show_progress: bool = False,
+) -> Iterator[bytes]:
+    """A generator that yields fixed-length chunks from binary_data.
+
+    Parameters
+    ----------
+    binary_data : Union[BinaryIO, socket.socket, bytes]
+        Binary data source.
+    packet_length_bytes : int
+        Number of bytes per packet to yield.
+    buffer_read_size_bytes : int, optional
+        Number of bytes to read from the source per read.
+    show_progress : bool
+        If True, prints a status bar.
+
+    Yields
+    ------
+    bytes
+        Fixed-length packet bytes.
+    """
+    n_bytes_parsed = 0  # Keep track of how many bytes we have parsed
+    n_packets_parsed = 0  # Keep track of how many packets we have parsed
+    read_buffer, total_length_bytes, read_bytes_from_source, buffer_read_size_bytes = _setup_binary_reader(
+        binary_data, buffer_read_size_bytes
+    )
+    current_pos = 0  # Keep track of where we are in the buffer
+    start_time = time.time_ns()
+    while True:
+        if total_length_bytes and n_bytes_parsed == total_length_bytes:
+            break
+        if show_progress:
+            _print_progress(
+                current_bytes=n_bytes_parsed,
+                total_bytes=total_length_bytes,
+                start_time_ns=start_time,
+                current_packets=n_packets_parsed,
+            )
+        if current_pos > 20_000_000:
+            # Only trim the buffer after 20 MB read to prevent modifying
+            # the bitstream and trimming after every packet
+            read_buffer = read_buffer[current_pos:]
+            current_pos = 0
+        while len(read_buffer) - current_pos < packet_length_bytes:
+            result = read_bytes_from_source(buffer_read_size_bytes)
+            if not result:
+                break
+            read_buffer += result
+        packet_bytes = read_buffer[current_pos : current_pos + packet_length_bytes]
+        current_pos += packet_length_bytes
+        n_packets_parsed += 1
+        n_bytes_parsed += packet_length_bytes
+        yield packet_bytes
+    if show_progress:
+        _print_progress(
+            current_bytes=n_bytes_parsed,
+            total_bytes=total_length_bytes,
+            start_time_ns=start_time,
+            current_packets=n_packets_parsed,
+            end="\n",
+            log=True,
+        )