space_packet_parser.definitions =============================== .. py:module:: space_packet_parser.definitions .. autoapi-nested-parse:: Module for parsing XTCE xml files to specify packet format Attributes ---------- .. autoapisummary:: 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 Exceptions ---------- .. autoapisummary:: space_packet_parser.definitions.UnrecognizedPacketTypeError Classes ------- .. autoapisummary:: space_packet_parser.definitions.XtcePacketDefinition Module Contents --------------- .. py:data:: logger .. py:data:: CcsdsPacketHeaderElement .. py:data:: CCSDS_HEADER_DEFINITION .. py:data:: CCSDS_HEADER_LENGTH_BYTES :value: 6 .. py:exception:: UnrecognizedPacketTypeError(*args, partial_data: dict = None) Bases: :py:obj:`Exception` Error raised when we can't figure out which kind of packet we are dealing with based on the header .. py:attribute:: partial_data .. py:class:: XtcePacketDefinition(xtce_document: Union[str, pathlib.Path, TextIO], *, ns: Optional[dict] = None) Object representation of the XTCE definition of a CCSDS packet object .. py:attribute:: _tag_to_type_template .. py:attribute:: _sequence_container_cache .. py:attribute:: _parameter_cache .. py:attribute:: _parameter_type_cache .. py:attribute:: tree .. py:attribute:: ns .. py:attribute:: type_tag_to_object .. py:method:: __getitem__(item) .. py:method:: _populate_sequence_container_cache() Force populating sequence_container_cache by parsing all SequenceContainers .. py:method:: 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. :param sequence_container: The SequenceContainer element to parse. :type sequence_container: ElementTree.Element :returns: SequenceContainer containing an entry_list of SequenceContainers and Parameters in the order expected in a packet. :rtype: SequenceContainer .. py:property:: named_containers :type: Dict[str, space_packet_parser.packets.SequenceContainer] Property accessor that returns the dict cache of SequenceContainer objects .. py:property:: named_parameters :type: Dict[str, space_packet_parser.parameters.Parameter] Property accessor that returns the dict cache of Parameter objects .. py:property:: named_parameter_types :type: Dict[str, space_packet_parser.parameters.ParameterType] Property accessor that returns the dict cache of ParameterType objects .. py:property:: container_set :type: lxml.etree.Element ContainerSet> element, containing all the sequence container elements. :type: Property that returns the element, containing all parameter type elements. :type: Property that returns the element, containing all parameter elements. :type: Property that returns the bool :staticmethod: Determine in a SequenceContainer element is abstract :param container_element: SequenceContainer element to examine :type container_element: ElementTree.Element :returns: True if SequenceContainer element has the attribute abstract=true. False otherwise. :rtype: bool .. py:method:: _find_container(name: str) -> lxml.etree.Element Finds an XTCE container by name. :param name: Name of the container to find :type name: str :rtype: ElementTree.Element .. py:method:: _find_parameter(name: str) -> lxml.etree.Element Finds an XTCE Parameter in the tree. :param name: Name of the parameter to find :type name: str :rtype: ElementTree.Element .. py:method:: _find_parameter_type(name: str) -> lxml.etree.Element Finds an XTCE ParameterType in the tree. :param name: Name of the parameter type to find :type name: str :rtype: ElementTree.Element .. py:method:: _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. :param container_element: The container element for which to find its base container. :type container_element: ElementTree.Element :returns: * *ElementTree.Element* -- The base container element of the input container_element. * *list* -- The restriction criteria for the inheritance. .. py:method:: _parse_header(packet_data: bytes) -> dict :staticmethod: Parses the CCSDS standard header. :param packet_data: 6 bytes of binary data. :type packet_data: bytes :returns: **header** -- Dictionary of header items. :rtype: dict .. py:method:: 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 :param packet: Binary representation of the packet used to get the coming bits and any previously parsed data items to infer field lengths. :type packet: packets.CCSDSPacket :param root_container_name: Default is CCSDSPacket. Any root container may be specified. :type root_container_name: str :returns: A Packet object containing header and data attributes. :rtype: Packet .. py:method:: _print_progress(*, current_bytes: int, total_bytes: Optional[int], start_time_ns: int, current_packets: int, end: str = '\r', log: bool = False) :staticmethod: Prints a progress bar, including statistics on parsing rate. :param current_bytes: Number of bytes parsed so far. :type current_bytes: int :param total_bytes: Number of total bytes to parse, if known. None otherwise. :type total_bytes: Optional[int] :param current_packets: Number of packets parsed so far. :type current_packets: int :param start_time_ns: Start time on system clock, in nanoseconds. :type start_time_ns: int :param end: Print function end string. Default is `\r` to create a dynamically updating loading bar. :type end: str :param log: If True, log the progress bar at INFO level. :type log: bool .. py:method:: packet_generator(binary_data: Union[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: Optional[int] = None, skip_header_bytes: int = 0) -> Iterator[Union[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. :param binary_data: Binary data source to parse into Packets. :type binary_data: Union[BinaryIO, socket.socket] :param parse_bad_pkts: 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. :type parse_bad_pkts: bool :param root_container_name: 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'. :type root_container_name: str :param ccsds_headers_only: Default False. If True, only parses the packet headers (does not use the provided packet definition). :type ccsds_headers_only: bool :param yield_unrecognized_packet_errors: 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. :type yield_unrecognized_packet_errors: bool :param show_progress: Default False. If True, prints a status bar. Note that for socket sources, the percentage will be zero until the generator ends. :type show_progress: bool :param buffer_read_size_bytes: 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. :type buffer_read_size_bytes: Optional[int] :param skip_header_bytes: 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. :type skip_header_bytes: int :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.