space_packet_parser.generators.utils

Utility functions for building packet generators.

Attributes

logger

Functions

`_read_packet_file`(→ Union[bytes, io.BufferedIOBase, ...)	Read a packet file or file-like object and return an object suitable for passing to a generator.
`_`(binary_data[, buffer_read_size_bytes])	Informative error if someone tries to pass a text file.
`_`(binary_data[, buffer_read_size_bytes])	Informative error if someone tries to pass a text file.
`_`(binary_data[, buffer_read_size_bytes])	Informative error if someone tries to pass a text file.
`_`(binary_data[, buffer_read_size_bytes])	Informative error if someone tries to pass a text file.
`_`(binary_data[, buffer_read_size_bytes])	Informative error if someone tries to pass a text file.
`_setup_binary_reader`(→ tuple)	Helper to set up reading from binary_data (file, socket, bytes) for a packet generator.
`_`(binary_data[, buffer_read_size_bytes])	Informative error if someone tries to pass a text file.
`_`(binary_data[, buffer_read_size_bytes])	Informative error if someone tries to pass a text file.
`_`(binary_data[, buffer_read_size_bytes])	Informative error if someone tries to pass a text file.
`_`(binary_data[, buffer_read_size_bytes])	Informative error if someone tries to pass a text file.
`_print_progress`(*, current_bytes, total_bytes, ...[, ...])	Prints a progress bar for a packet generator, including statistics on parsing rate.

Module Contents

space_packet_parser.generators.utils.logger

space_packet_parser.generators.utils._read_packet_file(packet_file) → bytes | io.BufferedIOBase | io.RawIOBase

Read a packet file or file-like object and return an object suitable for passing to a generator.

Specifically this function prepares the input for use with _setup_binary_reader.

Parameters:: packet_file (Union[str, Path, PathLike, BinaryIO, bytes])

Notes

This function handles strings and pathlike objects but it reads the full file into memory for the generator. This alleviates the need for generators to internally handle opening and closing files. For a more memory efficient approach, pass an opened file object.

space_packet_parser.generators.utils._(packet_file: io.BufferedIOBase | io.RawIOBase) → io.BufferedIOBase | io.RawIOBase: File-like object, this can be passed directly to a generator.

space_packet_parser.generators.utils._(packet_file: bytes) → bytes: bytes input, return as-is.

space_packet_parser.generators.utils._(packet_file: str) → bytes: String file path, open and read bytes.

space_packet_parser.generators.utils._(packet_file: pathlib.Path) → bytes

Path file path, open and read bytes.

Notes

This is a bit inefficient as we have to read the entire file into memory, but it does ensure we close the file after reading.

space_packet_parser.generators.utils._(packet_file: os.PathLike) → bytes

PathLike file path (e.g. anything supporting the Path interface), open and read bytes.

Notes

This is a bit inefficient as we have to read the entire file into memory, but it does ensure we close the file after reading.

space_packet_parser.generators.utils._setup_binary_reader(binary_data, buffer_read_size_bytes=None) → tuple

Helper to set up reading from binary_data (file, socket, bytes) for a packet generator.

Parameters:

binary_data (Union[io.BufferedIOBase, socket.socket, bytes]) – The binary data source to read from.
buffer_read_size_bytes (Optional[int]) – The number of bytes to read at a time from the source. If None, defaults to a sensible value based on the type of source.

Return type:

read_buffer, total_length_bytes, read_bytes_from_source, buffer_read_size_bytes

Notes

This function does not handle pathlike objects. It expects objects to be opened and readable already.

space_packet_parser.generators.utils._(binary_data: io.BufferedIOBase | io.RawIOBase, buffer_read_size_bytes=None) → tuple: Set up a binary reader from a file-like object.

space_packet_parser.generators.utils._(binary_data: socket.socket, buffer_read_size_bytes=None) → tuple: Set up a binary reader from a socket object.

space_packet_parser.generators.utils._(binary_data: bytes, buffer_read_size_bytes=None) → tuple: Set up a binary reader from a bytes object.

space_packet_parser.generators.utils._(binary_data: io.TextIOWrapper, buffer_read_size_bytes=None): Informative error if someone tries to pass a text file.

space_packet_parser.generators.utils._print_progress(*, current_bytes: int, total_bytes: int | None, start_time_ns: int, current_packets: int, end: str = '\r', log: bool = False)

Prints a progress bar for a packet generator, 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.