space_packet_parser.definitions

Module for parsing XTCE xml files to specify packet format

Attributes

`logger`
`CcsdsPacketHeaderElement`
`CCSDS_HEADER_DEFINITION`
`CCSDS_HEADER_LENGTH_BYTES`

Exceptions

UnrecognizedPacketTypeError

Error raised when we can't figure out which kind of packet we are dealing with based on the header

Classes

XtcePacketDefinition

Object representation of the XTCE definition of a CCSDS packet object

Module Contents

space_packet_parser.definitions.logger

space_packet_parser.definitions.CcsdsPacketHeaderElement

space_packet_parser.definitions.CCSDS_HEADER_DEFINITION

space_packet_parser.definitions.CCSDS_HEADER_LENGTH_BYTES = 6

exception space_packet_parser.definitions.UnrecognizedPacketTypeError(*args, partial_data: dict = None)

Bases: Exception

Error raised when we can’t figure out which kind of packet we are dealing with based on the header

partial_data

class space_packet_parser.definitions.XtcePacketDefinition(xtce_document: str | pathlib.Path | TextIO, *, ns: dict | None = None)

Object representation of the XTCE definition of a CCSDS packet object

_tag_to_type_template

_sequence_container_cache

_parameter_cache

_parameter_type_cache

tree

ns

type_tag_to_object

__getitem__(item)

_populate_sequence_container_cache(): Force populating sequence_container_cache by parsing all SequenceContainers

parse_sequence_container_contents(sequence_container: lxml.etree.Element) → space_packet_parser.packets.SequenceContainer

Parses the list of parameters in a SequenceContainer element, recursively parsing nested SequenceContainers to build an entry list of parameters that flattens the nested structure to derive a sequential ordering of expected parameters for each SequenceContainer. Note that this also stores entry lists for containers that are not intended to stand alone.

Parameters:: sequence_container (ElementTree.Element) – The SequenceContainer element to parse.
Returns:: SequenceContainer containing an entry_list of SequenceContainers and Parameters in the order expected in a packet.
Return type:: SequenceContainer

property named_containers: Dict[str, space_packet_parser.packets.SequenceContainer]: Property accessor that returns the dict cache of SequenceContainer objects

property named_parameters: Dict[str, space_packet_parser.parameters.Parameter]: Property accessor that returns the dict cache of Parameter objects

property named_parameter_types: Dict[str, space_packet_parser.parameters.ParameterType]: Property accessor that returns the dict cache of ParameterType objects

property container_set: lxml.etree.Element

ContainerSet> element, containing all the sequence container elements.

Type:: Property that returns the <xtce

property parameter_type_set: lxml.etree.Element

ParameterTypeSet> element, containing all parameter type elements.

Type:: Property that returns the <xtce

property parameter_set: lxml.etree.Element

ParameterSet> element, containing all parameter elements.

Type:: Property that returns the <xtce

static _is_abstract_container(container_element: lxml.etree.Element) → bool

Determine in a SequenceContainer element is abstract

Parameters:: container_element (ElementTree.Element) – SequenceContainer element to examine
Returns:: True if SequenceContainer element has the attribute abstract=true. False otherwise.
Return type:: bool

_find_container(name: str) → lxml.etree.Element

Finds an XTCE container <xtce:SequenceContainer> by name.

Parameters:: name (str) – Name of the container to find
Return type:: ElementTree.Element

_find_parameter(name: str) → lxml.etree.Element

Finds an XTCE Parameter in the tree.

Parameters:: name (str) – Name of the parameter to find
Return type:: ElementTree.Element

_find_parameter_type(name: str) → lxml.etree.Element

Finds an XTCE ParameterType in the tree.

Parameters:: name (str) – Name of the parameter type to find
Return type:: ElementTree.Element

_get_container_base_container(container_element: lxml.etree.Element) → Tuple[lxml.etree.Element, List[space_packet_parser.comparisons.MatchCriteria]]

Examines the container_element and returns information about its inheritance.

Parameters:

container_element (ElementTree.Element) – The container element for which to find its base container.

Returns:

ElementTree.Element – The base container element of the input container_element.
list – The restriction criteria for the inheritance.

static _parse_header(packet_data: bytes) → dict

Parses the CCSDS standard header.

Parameters:: packet_data (bytes) – 6 bytes of binary data.
Returns:: header – Dictionary of header items.
Return type:: dict

parse_ccsds_packet(packet: space_packet_parser.packets.CCSDSPacket, *, root_container_name: str = 'CCSDSPacket', **parse_value_kwargs) → space_packet_parser.packets.CCSDSPacket

Parse binary packet data according to the self.packet_definition object

Parameters:

packet (packets.CCSDSPacket) – Binary representation of the packet used to get the coming bits and any previously parsed data items to infer field lengths.
root_container_name (str) – Default is CCSDSPacket. Any root container may be specified.

Returns:

A Packet object containing header and data attributes.

Return type:

Packet

static _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, 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.

packet_generator(binary_data: BinaryIO | socket.socket, *, parse_bad_pkts: bool = True, root_container_name='CCSDSPacket', ccsds_headers_only: bool = False, yield_unrecognized_packet_errors: bool = False, show_progress: bool = False, buffer_read_size_bytes: int | None = None, skip_header_bytes: int = 0) → Iterator[space_packet_parser.packets.CCSDSPacket | UnrecognizedPacketTypeError]

Create and return a Packet generator that reads from a ConstBitStream or a filelike object or a socket.

Creating a generator object to return allows the user to create many generators from a single Parser and reduces memory usage.

Parameters:

binary_data (Union[BinaryIO, socket.socket]) – Binary data source to parse into Packets.
parse_bad_pkts (bool) – Default True. If True, when the generator encounters a packet with an incorrect length it will still yield the packet (the data will likely be invalid). If False, the generator will still write a debug log message but will otherwise silently skip the bad packet.
root_container_name (str) – The name of the root level (lowest level of container inheritance) SequenceContainer. This SequenceContainer is assumed to be inherited by every possible packet structure in the XTCE document and is the starting point for parsing. Default is ‘CCSDSPacket’.
ccsds_headers_only (bool) – Default False. If True, only parses the packet headers (does not use the provided packet definition).
yield_unrecognized_packet_errors (bool) – Default False. If False, UnrecognizedPacketTypeErrors are caught silently and parsing continues to the next packet. If True, the generator will yield an UnrecognizedPacketTypeError in the event of an unrecognized packet. Note: These exceptions are not raised by default but are instead returned so that the generator can continue. You can raise the exceptions if desired. Leave this as False unless you need to examine the partial data from unrecognized packets.
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.
buffer_read_size_bytes (Optional[int]) – 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.
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.

Yields:

Union[Packet, UnrecognizedPacketTypeError] – Generator yields Packet objects containing the parsed packet data for each subsequent packet. If yield_unrecognized_packet_errors is True, it will yield an unraised exception object, which can be raised or used for debugging purposes.