space_packet_parser =================== .. py:module:: space_packet_parser .. autoapi-nested-parse:: Space Packet Parser Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/space_packet_parser/cli/index /autoapi/space_packet_parser/common/index /autoapi/space_packet_parser/definitions/index /autoapi/space_packet_parser/exceptions/index /autoapi/space_packet_parser/generators/index /autoapi/space_packet_parser/packets/index /autoapi/space_packet_parser/xarr/index /autoapi/space_packet_parser/xtce/index Classes ------- .. autoapisummary:: space_packet_parser.SpacePacket space_packet_parser.XtcePacketDefinition Functions --------- .. autoapisummary:: space_packet_parser.ccsds_generator space_packet_parser.validate_xtce space_packet_parser.load_xtce Package Contents ---------------- .. py:class:: SpacePacket(*args, binary_data: bytes = b'', **kwargs) Bases: :py:obj:`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. :param \*args: Initial items to store in the packet, passed to the dict() constructor. :type \*args: Mapping or Iterable :param binary_data: 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. :type binary_data: bytes, optional :param \*\*kwargs: Additional packet items to store, passed to the dict() constructor. :type \*\*kwargs: dict .. py:attribute:: binary_data :value: b'' .. py:attribute:: _parsing_pos :value: 0 .. py:property:: header :type: dict The header content of the packet. .. py:property:: user_data :type: dict The user data content of the packet. .. py:property:: raw_data :type: bytes The raw binary data of the packet. .. py:method:: _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. :param nbits: Number of bits to read :type nbits: int :returns: Raw bytes from the packet data :rtype: bytes .. py:method:: _read_from_binary_as_int(nbits: int) -> int Read a number of bits from the binary packet data as an integer. :param nbits: Number of bits to read :type nbits: int :returns: Integer representation of the bits read from the packet :rtype: int .. py:function:: 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. :param binary_data: Binary data source containing CCSDSPackets. :type binary_data: Union[io.BufferedIOBase, io.RawIOBase, socket.socket, bytes] :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: int, optional :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 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 :param combine_segmented_packets: 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. :type combine_segmented_packets: bool :param secondary_header_bytes: 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, ...]. :type secondary_header_bytes: int :Yields: *CCSDSPacketBytes* -- The bytes of a single CCSDS packet. .. py:class:: 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: :py:obj:`space_packet_parser.common.AttrComparable` Object representation of the XTCE definition of a CCSDS packet object .. py:attribute:: parameter_types :type: dict[str, space_packet_parser.xtce.parameter_types.ParameterType] .. py:attribute:: parameters :type: dict[str, space_packet_parser.xtce.parameters.Parameter] .. py:attribute:: containers :type: dict[str, space_packet_parser.xtce.containers.SequenceContainer] .. py:attribute:: ns .. py:attribute:: xtce_ns_uri :value: 'https://www.omg.org/spec/XTCE/20180204' .. py:attribute:: xtce_ns_prefix :value: 'xtce' .. py:attribute:: root_container_name :value: 'CCSDSPacket' .. py:attribute:: space_system_name :value: None .. py:attribute:: validation_status :value: 'Unknown' .. py:attribute:: xtce_version :value: '1.0' .. py:attribute:: date :value: None .. py:method:: write_xml(filepath: str | pathlib.Path) -> None Write out the XTCE XML for this packet definition object to the specified path :param filepath: Location to write this packet definition :type filepath: Union[str, Path] .. py:method:: to_xml_tree() -> lxml.etree.ElementTree Initializes and returns an ElementTree object based on parameter type, parameter, and container information :rtype: ElementTree.ElementTree .. py:method:: from_xtce(xtce_document: str | pathlib.Path | os.PathLike | TextIO, *, root_container_name: str = DEFAULT_ROOT_CONTAINER) -> XtcePacketDefinition :classmethod: .. py:method:: _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] :staticmethod: Parse the element into a dictionary of SequenceContainer objects :param tree: Full XTCE tree :type tree: ElementTree.Element :param parameter_lookup: Parameters that are contained in container entry lists :type parameter_lookup: dict[str, parameters.Parameter] :rtype: dict[str, containers.SequenceContainer] .. py:method:: _parse_parameter_type_set(tree: lxml.etree.ElementTree) -> dict[str, space_packet_parser.xtce.parameter_types.ParameterType] :staticmethod: Parse the into a dictionary of ParameterType objects :param tree: Full XTCE tree :type tree: ElementTree.ElementTree :rtype: dict[str, parameters.ParameterType] .. py:method:: _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] :staticmethod: Parse an object into a dictionary of Parameter objects :param tree: Full XTCE tree :type tree: ElementTree.ElementTree :param parameter_type_lookup: Parameter types referenced by parameters. :type parameter_type_lookup: dict[str, parameter_types.ParameterType] :rtype: dict[str, parameters.Parameter] .. py:method:: 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 :param binary_data: Binary representation of the packet used to get the coming bits and any previously parsed data items to infer field lengths. :type binary_data: bytes :param root_container_name: 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. :type root_container_name: Optional[str] :returns: A SpacePacket object containing header and data attributes. :rtype: SpacePacket .. py:method:: 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 :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: space_packet_parser.SpacePacket :param root_container_name: 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. :type root_container_name: Optional[str] :returns: A SpacePacket object containing header and data attributes. :rtype: SpacePacket .. py:method:: 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 :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: space_packet_parser.SpacePacket :param root_container_name: 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. :type root_container_name: Optional[str] :returns: A SpacePacket object containing header and data attributes. :rtype: SpacePacket .. py:function:: 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. :param xml_source: Path to XML file, XML string content, or ElementTree :type xml_source: Union[str, Path, ElementTree.ElementTree] :param level: Validation level: "schema", "structure", or "all". Default "all". :type level: str :param timeout: Timeout in seconds for schema downloads :type timeout: int :param print_results: Default True. Prints results before returning Truthy or Falsy result. :type print_results: bool :param raise_on_error: Default True. If False, returns a ValidationResult object with information about the validation results. If True, raises an exception unless the ValidationResult reports valid. :type raise_on_error: bool :param local_xsd: 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. :type local_xsd: Optional[str, Path] :returns: Truthy if result is valid, Falsy otherwise :rtype: ValidationResult .. py:function:: 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(). :param filename: XTCE XML file :type filename: Union[str, Path] :rtype: definitions.XtcePacketDefinition