space_packet_parser

Space Packet Parser

Submodules

• space_packet_parser.cli
• space_packet_parser.common
• space_packet_parser.definitions
• space_packet_parser.exceptions
• space_packet_parser.generators
• space_packet_parser.packets
• space_packet_parser.xarr
• space_packet_parser.xtce

Classes

SpacePacket	Packet representing parsed data items.
XtcePacketDefinition	Object representation of the XTCE definition of a CCSDS packet object

Functions

ccsds_generator(...)	A generator that reads raw packet data from a filelike object or a socket.
validate_xtce(→ ValidationResult)	Validate an XTCE XML document.
load_xtce(→ xtce.definitions.XtcePacketDefinition)	Create an XtcePacketDefinition object from an XTCE XML file

Package Contents

class space_packet_parser.SpacePacket(*args, binary_data: bytes = b'', **kwargs)

Bases: dict

Packet representing parsed data items.

Container that stores the binary packet data (bytes) as an instance attribute and the parsed data items in a dictionary interface. A SpacePacket generally begins as an empty dictionary that gets filled as the packet is parsed. To access the raw bytes of the packet, use the SpacePacket.binary_data attribute.

Parameters:

• *args (Mapping or Iterable) – Initial items to store in the packet, passed to the dict() constructor.

• binary_data (bytes, optional) – The binary data for a single packet as a bytes object / subclass. This binary data is stored and used for parsing the data items. Internally we are tracking the parsing position within this binary_data object and trying to read specific bit ranges from it.

• **kwargs (dict) – Additional packet items to store, passed to the dict() constructor.

binary_data = b''

_parsing_pos = 0

property header: dict

The header content of the packet.

property user_data: dict

The user data content of the packet.

property raw_data: bytes

The raw binary data of the packet.

_read_from_binary_as_bytes(nbits: int) → bytes

Read a number of bits from the binary packet data as bytes.

Reads the minimum number of complete bytes required to capture nbits. Moves _parsing_pos cursor nbits forward, even if nbits is not an integer number of bytes.

Parameters:

nbits (int) – Number of bits to read

Returns:

Raw bytes from the packet data

Return type:

bytes

_read_from_binary_as_int(nbits: int) → int

Read a number of bits from the binary packet data as an integer.

Parameters:

nbits (int) – Number of bits to read

Returns:

Integer representation of the bits read from the packet

Return type:

int

space_packet_parser.ccsds_generator(binary_data: io.BufferedIOBase | io.RawIOBase | socket.socket | bytes, *, buffer_read_size_bytes: int | None = None, show_progress: bool = False, skip_header_bytes: int = 0, combine_segmented_packets: bool = False, secondary_header_bytes: int = 0) → collections.abc.Iterator[CCSDSPacketBytes]

A generator that reads raw packet data from a filelike object or a socket.

Each iteration of the generator yields a CCSDSPacketBytes object that makes up a single CCSDS packet. If combining segmented packets is enabled, the generator will combine segmented packets into a single packet for parsing. This is useful for parsing packets that are split into multiple packets due to size constraints.

Parameters:

• binary_data (Union[io.BufferedIOBase, io.RawIOBase, socket.socket, bytes]) – Binary data source containing CCSDSPackets.

• buffer_read_size_bytes (int, optional) – Number of bytes to read from e.g. a BufferedReader or socket binary data source on each read attempt. If None, defaults to 4096 bytes from a socket, -1 (full read) from a file.

• show_progress (bool) – Default False. If True, prints a status bar. Note that for socket sources, the percentage will be zero until the generator ends.

• skip_header_bytes (int) – Default 0. The parser skips this many bytes at the beginning of every packet. This allows dynamic stripping of additional header data that may be prepended to packets in “raw record” file formats.

• combine_segmented_packets (bool) – Default False. If True, combines segmented packets into a single packet for parsing. This is useful for parsing packets that are split into multiple packets due to size constraints. The packet data is combined by concatenating the data from each packet together. The combined packet is then parsed as a single packet. Only the first CCSDS header is kept when concatenating continued packets. CCSDS headers (and secondary headers) from continuation packets are discarded.

• secondary_header_bytes (int) – Default 0. The length of the secondary header in bytes. This is used to skip the secondary header of segmented packets. The byte layout within the returned packet has all data concatenated together as follows: [packet0header, packet0secondaryheader, packet0data, packet1data, packet2data, …].

Yields:

CCSDSPacketBytes – The bytes of a single CCSDS packet.

class space_packet_parser.XtcePacketDefinition(container_set: collections.abc.Iterable[space_packet_parser.xtce.containers.SequenceContainer] | None = None, *, ns: dict = STANDARD_XTCE_NSMAP, xtce_ns_prefix: str = STANDARD_XTCE_NS_PREFIX, root_container_name: str = DEFAULT_ROOT_CONTAINER, space_system_name: str | None = None, validation_status: str = 'Unknown', xtce_version: str = '1.0', date: str | None = None)

Bases: space_packet_parser.common.AttrComparable

Object representation of the XTCE definition of a CCSDS packet object

parameter_types: dict[str, space_packet_parser.xtce.parameter_types.ParameterType]

parameters: dict[str, space_packet_parser.xtce.parameters.Parameter]

containers: dict[str, space_packet_parser.xtce.containers.SequenceContainer]

ns

xtce_ns_uri = 'https://www.omg.org/spec/XTCE/20180204'

xtce_ns_prefix = 'xtce'

root_container_name = 'CCSDSPacket'

space_system_name = None

validation_status = 'Unknown'

xtce_version = '1.0'

date = None

write_xml(filepath: str | pathlib.Path) → None

Write out the XTCE XML for this packet definition object to the specified path

Parameters:

filepath (Union[str, Path]) – Location to write this packet definition

to_xml_tree() → lxml.etree.ElementTree

Initializes and returns an ElementTree object based on parameter type, parameter, and container information

Return type:

ElementTree.ElementTree

classmethod from_xtce(xtce_document: str | pathlib.Path | os.PathLike | TextIO, *, root_container_name: str = DEFAULT_ROOT_CONTAINER) → XtcePacketDefinition

static _parse_container_set(tree: lxml.etree.Element, parameter_lookup: dict[str, space_packet_parser.xtce.parameters.Parameter]) → dict[str, space_packet_parser.xtce.containers.SequenceContainer]

Parse the <xtce:ContainerSet> element into a dictionary of SequenceContainer objects

Parameters:

• tree (ElementTree.Element) – Full XTCE tree

• parameter_lookup (dict[str, parameters.Parameter]) – Parameters that are contained in container entry lists

Return type:

dict[str, containers.SequenceContainer]

static _parse_parameter_type_set(tree: lxml.etree.ElementTree) → dict[str, space_packet_parser.xtce.parameter_types.ParameterType]

Parse the <xtce:ParameterTypeSet> into a dictionary of ParameterType objects

Parameters:

tree (ElementTree.ElementTree) – Full XTCE tree

Return type:

dict[str, parameters.ParameterType]

static _parse_parameter_set(tree: lxml.etree.ElementTree, parameter_type_lookup: dict[str, space_packet_parser.xtce.parameter_types.ParameterType]) → dict[str, space_packet_parser.xtce.parameters.Parameter]

Parse an <xtce:ParameterSet> object into a dictionary of Parameter objects

Parameters:

• tree (ElementTree.ElementTree) – Full XTCE tree

• parameter_type_lookup (dict[str, parameter_types.ParameterType]) – Parameter types referenced by parameters.

Return type:

dict[str, parameters.Parameter]

parse_bytes(binary_data: bytes, *, root_container_name: str | None = None) → space_packet_parser.SpacePacket

Parse binary packet data according to the self.packet_definition object

Parameters:

• binary_data (bytes) – Binary representation of the packet used to get the coming bits and any previously parsed data items to infer field lengths.

• root_container_name (Optional[str]) – Default is taken from the XtcePacketDefinition object. Any root container may be specified, but it must begin with the definition of a CCSDS header in order to parse correctly.

Returns:

A SpacePacket object containing header and data attributes.

Return type:

SpacePacket

parse_packet(packet: space_packet_parser.SpacePacket, *, root_container_name: str | None = None) → space_packet_parser.SpacePacket

Parse binary packet data according to the self.packet_definition object

Parameters:

• packet (space_packet_parser.SpacePacket) – Binary representation of the packet used to get the coming bits and any previously parsed data items to infer field lengths.

• root_container_name (Optional[str]) – Default is taken from the XtcePacketDefinition object. Any root container may be specified, but it must begin with the definition of a CCSDS header in order to parse correctly.

Returns:

A SpacePacket object containing header and data attributes.

Return type:

SpacePacket

parse_ccsds_packet(packet: space_packet_parser.SpacePacket, *, root_container_name: str | None = None) → space_packet_parser.SpacePacket

Parse binary packet data according to the self.packet_definition object

Parameters:

• packet (space_packet_parser.SpacePacket) – Binary representation of the packet used to get the coming bits and any previously parsed data items to infer field lengths.

• root_container_name (Optional[str]) – Default is taken from the XtcePacketDefinition object. Any root container may be specified, but it must begin with the definition of a CCSDS header in order to parse correctly.

Returns:

A SpacePacket object containing header and data attributes.

Return type:

SpacePacket

space_packet_parser.validate_xtce(xml_source: str | pathlib.Path | lxml.etree.ElementTree, level: str = 'all', timeout: int = 30, print_results: bool = True, raise_on_error: bool = True, local_xsd: str | pathlib.Path | None = None) → ValidationResult

Validate an XTCE XML document.

This is the main validation entry point for XTCE documents. It can perform schema or structural validation based on the level parameter.

Parameters:

• xml_source (Union[str, Path, ElementTree.ElementTree]) – Path to XML file, XML string content, or ElementTree

• level (str) – Validation level: “schema”, “structure”, or “all”. Default “all”.

• timeout (int) – Timeout in seconds for schema downloads

• print_results (bool) – Default True. Prints results before returning Truthy or Falsy result.

• raise_on_error (bool) – Default True. If False, returns a ValidationResult object with information about the validation results. If True, raises an exception unless the ValidationResult reports valid.

• local_xsd (Optional[str, Path]) – Local path to an XSD for schema validation. If not provided and schema validation is requested, XSD is retrieved from schema reference attribute in document root.

Returns:

Truthy if result is valid, Falsy otherwise

Return type:

ValidationResult

space_packet_parser.load_xtce(filename: str | pathlib.Path) → xtce.definitions.XtcePacketDefinition

Create an XtcePacketDefinition object from an XTCE XML file

This is a shortcut for calling XtcePacketDefinition.from_xtce().

Parameters:

filename (Union[str, Path]) – XTCE XML file

Return type:

definitions.XtcePacketDefinition