epicsarchiver.retrieval.pb

Handle retrieval of data in the Archiver Appliance PB file format.

The file format is described on this page: https://slacmshankar.github.io/epicsarchiver_docs/pb_pbraw.html

The data can be parsed in the same way whether retrieved using the Rest API or whether reading files directly from disk. In either case, it is important to treat the data as binary data - a stream of bytes. The Google Protobuf library handles converting the stream of bytes into the objects defined by the EPICSEvent.proto file.

The Archiver Appliance escapes certain characters as described on the page above, which allows one to deduce the number of events in the binary file using tools such as wc.

The unescape_bytes() method handles unescaping these characters before handing the interpretation over to the Google Protobuf library.

Attributes

`LOG`
`TYPE_MAPPINGS`
`INVERSE_TYPE_MAPPINGS`
`ESC_BYTE`
`NL_BYTE`
`CR_BYTE`
`PB_REPLACEMENTS_ESCAPING`
`PB_REPLACEMENTS_UNESCAPING`
`EeScalarEvent`
`EeVectorEvent`
`EeEvent`
`RE_ESCAPE_PATTERN`

Functions

`unescape_bytes`(→ bytes)	Replace specific sub-sequences in a bytes sequence.
`escape_bytes`(→ bytes)	Replace specific sub-sequences in a bytes sequence.
`event_timestamp`(→ datetime.datetime)	Converts from protobuf event time format to python datetime.
`get_timestamp_from_line_function`(...)	From a unescaped protobuf line create a function to get datetime.
`_break_up_chunks`(→ collections.OrderedDict[int, ...)	Break up raw data into chunks by year.
`_event_from_line`(...)	Get an ArchiveEvent from this line.
`_try_recover_truncated_event`(→ EeEvent \| None)	Try to recover an event truncated by an unescaped 0x0a byte (archiver bug).
`parse_pb_data`(...)	Turn raw PB data into an ArchiveData object.
`metadata_from_chunk_info`(...)	Convert a chunk info into metadata.
`get_iso_timestamp_for_event`(→ str)	Returns an ISO-formatted timestamp string for the given event.
`split_pb_chunks`(...)	Split raw endpoint data into per-year (PayloadInfo, chunk_bytes) pairs.
`read_pb_file`(...)	Read an unescaped protobuf file and produce a list of events from file.
`to_field_value`(...)	From the protobuf variant.

Module Contents

epicsarchiver.retrieval.pb.LOG: logging.Logger[source]

epicsarchiver.retrieval.pb.TYPE_MAPPINGS: dict[int, type][source]

epicsarchiver.retrieval.pb.INVERSE_TYPE_MAPPINGS[source]

epicsarchiver.retrieval.pb.ESC_BYTE = b'\x1b'[source]

epicsarchiver.retrieval.pb.NL_BYTE = b'\n'[source]

epicsarchiver.retrieval.pb.CR_BYTE = b'\r'[source]

epicsarchiver.retrieval.pb.PB_REPLACEMENTS_ESCAPING[source]

epicsarchiver.retrieval.pb.PB_REPLACEMENTS_UNESCAPING[source]

type epicsarchiver.retrieval.pb.EeScalarEvent = ee.ScalarString | ee.ScalarShort | ee.ScalarFloat | ee.ScalarEnum | ee.ScalarByte | ee.ScalarInt | ee.ScalarDouble[source]

epicsarchiver.retrieval.pb.EeVectorEvent[source]

type epicsarchiver.retrieval.pb.EeEvent = EeScalarEvent | EeVectorEvent[source]

epicsarchiver.retrieval.pb.RE_ESCAPE_PATTERN[source]

epicsarchiver.retrieval.pb.unescape_bytes(byte_seq: bytes) → bytes[source]

Replace specific sub-sequences in a bytes sequence.

This escaping is defined as part of the Archiver Appliance raw file format: https://slacmshankar.github.io/epicsarchiver_docs/pb_pbraw.html

Parameters:: byte_seq – any byte sequence
Returns:: the byte sequence unescaped according to the AA file format rules

epicsarchiver.retrieval.pb.escape_bytes(byte_seq: bytes) → bytes[source]

Replace specific sub-sequences in a bytes sequence.

This escaping is defined as part of the Archiver Appliance raw file format: https://slacmshankar.github.io/epicsarchiver_docs/pb_pbraw.html

Parameters:: byte_seq – any byte sequence
Returns:: the byte sequence escaped according to the AA file format rules

epicsarchiver.retrieval.pb.event_timestamp(year: int, event: EeEvent) → datetime.datetime[source]

Converts from protobuf event time format to python datetime.

Parameters:

year (int) – year of event
event (EeEvent) – input event

Returns:

Output datetime

Return type:

pydt

epicsarchiver.retrieval.pb.get_timestamp_from_line_function(chunk_info: epicsarchiver.retrieval.EPICSEvent_pb2.PayloadInfo) → collections.abc.Callable[[bytes], datetime.datetime][source]

From a unescaped protobuf line create a function to get datetime.

Parameters:: chunk_info (ee.PayloadInfo) – Payload info of protobuf file
Returns:: Function to provide event time
Return type:: Callable[[bytes], pydt]

epicsarchiver.retrieval.pb._break_up_chunks(raw_data: bytes) → collections.OrderedDict[int, tuple[epicsarchiver.retrieval.EPICSEvent_pb2.PayloadInfo, list[bytes]]][source]

Break up raw data into chunks by year.

Parameters:: raw_data – Raw data from file
Returns:: keys are years; values are lists of chunks
Return type:: collections.OrderedDict

epicsarchiver.retrieval.pb._event_from_line(line: bytes, pv: str, year: int, event_type: int, line_number: int = 0) → epicsarchiver.retrieval.archive_event.ArchiveEvent | None[source]

Get an ArchiveEvent from this line.

Parameters:

line – A line of chunks of data
pv – Name of the PV
year – Year of interest
event_type – Need to know the type of the event as key of TYPE_MAPPINGS
line_number – Line number in the file

Returns:

The event

Return type:

ArchiveEvent

epicsarchiver.retrieval.pb._try_recover_truncated_event(event_type: int, unescaped: bytes, line_number: int, pv: str) → EeEvent | None[source]

Try to recover an event truncated by an unescaped 0x0a byte (archiver bug).

When a protobuf varint value byte equals 0x0a, split(b”n”) truncates the event there, leaving the orphaned field tag as the last byte. We detect any varint-wire-type tag at end-of-line and retry with NL_BYTE appended.

Returns:: The recovered event on success, None if not applicable or recovery fails.

epicsarchiver.retrieval.pb.parse_pb_data(raw_data: bytes) → epicsarchiver.retrieval.archive_event.ArchiveEventsData[source]

Turn raw PB data into an ArchiveData object.

Parameters:: raw_data – The raw data
Returns:: An ArchiveData object

epicsarchiver.retrieval.pb.metadata_from_chunk_info(year: int, chunk_info: epicsarchiver.retrieval.EPICSEvent_pb2.PayloadInfo) → epicsarchiver.retrieval.archive_event.ArchiveEventsMeta[source]

Convert a chunk info into metadata.

Parameters:

year (int) – Year of interest
chunk_info (ee.PayloadInfo) – Input chunk info

Returns:

Output metadata

Return type:

ArchiveEventsMeta

epicsarchiver.retrieval.pb.get_iso_timestamp_for_event(year: int, event: EeEvent) → str[source]: Returns an ISO-formatted timestamp string for the given event.

epicsarchiver.retrieval.pb.split_pb_chunks(raw_data: bytes) → list[tuple[epicsarchiver.retrieval.EPICSEvent_pb2.PayloadInfo, bytes]][source]

Split raw endpoint data into per-year (PayloadInfo, chunk_bytes) pairs.

The archiver endpoint may return data spanning multiple years as chunks separated by double newlines. Each returned chunk is a self-contained PB file: an escaped PayloadInfo line followed by escaped event lines.

Parameters:: raw_data – Raw bytes from the archiver getData endpoint
Returns:: List of (PayloadInfo, chunk_bytes) tuples, one per year

epicsarchiver.retrieval.pb.read_pb_file(filename: str) → epicsarchiver.retrieval.archive_event.ArchiveEventsData[source]

Read an unescaped protobuf file and produce a list of events from file.

Parameters:: filename (str) – location of file
Returns:: list of events in file
Return type:: list[ArchiveEvent]

epicsarchiver.retrieval.pb.to_field_value(f: epicsarchiver.retrieval.EPICSEvent_pb2.FieldValue) → epicsarchiver.retrieval.archive_event.FieldValue[source]

From the protobuf variant.

Parameters:: f (ee.FieldValue) – protobuf field value
Returns:: Basic Field Value
Return type:: FieldValue