space_packet_parser.xarr

Extras package that supports generating an xarray.Dataset directly

Attributes

logger

ReadableBinaryPacket

Functions

_min_dtype_for_encoding(data_encoding)

Find the minimum data type capable of representing an XTCE data encoding.

_get_minimum_numpy_datatype(→ str | None)

Get the minimum datatype for a given variable.

create_dataset(→ dict[int, xarray.Dataset])

Create a dictionary of xarray Datasets (per APID) from a set of packet files

Module Contents

space_packet_parser.xarr.logger
space_packet_parser.xarr.ReadableBinaryPacket
space_packet_parser.xarr._min_dtype_for_encoding(data_encoding: space_packet_parser.xtce.encodings.DataEncoding)

Find the minimum data type capable of representing an XTCE data encoding.

This only works for raw values and does not apply to calibrated or otherwise derived values.

Parameters:

data_encoding (encodings.DataEncoding) – The raw data encoding.

Returns:

The numpy dtype string for the minimal representation of the data encoding.

Return type:

str

space_packet_parser.xarr._get_minimum_numpy_datatype(name: str, definition: space_packet_parser.xtce.definitions.XtcePacketDefinition, use_raw_value: bool = False) str | None

Get the minimum datatype for a given variable.

Parameters:
  • name (str) – The variable name.

  • definition (xtce.definitions.XtcePacketDefinition) – The XTCE packet definition. Used to examine data types to infer their niminal numpy representation.

  • use_raw_value (bool) – Default False. If True, uses the data type of the raw value for each parameter.

Returns:

datatype – The minimum numpy dtype for the parameter. Returns None to indicate that numpy should use default dtype inference.

Return type:

Optional[str]

space_packet_parser.xarr.create_dataset(packet_files: ReadableBinaryPacket | collections.abc.Iterable[ReadableBinaryPacket], xtce_packet_definition: str | os.PathLike | space_packet_parser.xtce.definitions.XtcePacketDefinition, use_raw_values: bool = False, packet_bytes_generator: collections.abc.Callable | None = None, generator_kwargs: dict | None = None, parse_bytes_kwargs: dict | None = None, packet_filter: collections.abc.Callable[[bytes], bool] | None = None) dict[int, xarray.Dataset]

Create a dictionary of xarray Datasets (per APID) from a set of packet files

Notes

This function only handles packet definitions with the same variable structure across all packets with the same APID. For example, this cannot be used for polymorphic packets whose structure changes based on previously parsed values. If you are parsing muxed APID data and wish to filter it, you can either omit the unwanted APID packet definitions from your XTCE (fastest) or you can parse everything and filter after creation of the Dataset for each APID.

Parameters:
  • packet_files (Union[str, Path, PathLike, io.BufferedIOBase, io.RawIOBase, bytes, Iterable[Union[str, Path, PathLike, io.BufferedIOBase, io.RawIOBase, bytes]]]) – Packet files or file-like objects opened in binary mode. Accepts file paths (str, Path, PathLike), file-like objects (io.BufferedIOBase, io.RawIOBase), or raw bytes.

  • xtce_packet_definition (Union[str, Path, PathLike, xtce.definitions.XtcePacketDefinition]) – Packet definition for parsing the packet data

  • use_raw_values (bool) – Default False. If True, saves parameter raw values to the resulting Dataset. e.g. enumerated lookups will be saved as their encoded integer values.

  • packet_bytes_generator (Optional[callable]) – The generator function to use for yielding packet bytes. Defaults to ccsds.ccsds_generator. Can be set to fixed_length_generator or any other generator that yields bytes-like objects.

  • generator_kwargs (Optional[dict]) – Keyword arguments passed to the packet bytes generator.

  • parse_bytes_kwargs (Optional[dict]) – Keyword arguments passed to XtcePacketDefinition.parse_bytes().

  • packet_filter (Optional[Callable[[bytes], bool]]) – Optional function to filter items returned from the packet generator (e.g. lambda pkt: pkt.apid == 100)

Returns:

Dataset objects parsed from the iterable of packets, keyed by APID.

Return type:

dict[int, xr.Dataset]