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.

Note: due to the way the protobuf objects are constructed, pylint can’t correctly deduce some properties, so I have manually disabled some warnings.

Attributes

LOG
TYPE_MAPPINGS
INVERSE_TYPE_MAPPINGS
ESC_BYTE
NL_BYTE
CR_BYTE
PB_REPLACEMENTS_ESCAPING
PB_REPLACEMENTS_UNESCAPING
EeScalarEvent
EeVectorEvent
EeEvent

Functions

unescape_bytes(→ bytes)	Replace specific sub-sequences in a bytes sequence.
escape_bytes(→ bytes)	Replace specific sub-sequences in a bytes sequence.
event_pd_timestamp(→ pandas.Timestamp)	Converts from protobuf event time format to python datetime.
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.
parse_pb_data(...)	Turn raw PB data into an ArchiveData object.
get_iso_timestamp_for_event(→ str)	Returns an ISO-formatted timestamp string for the given event.
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

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

epicsarchiver.retrieval.pb.INVERSE_TYPE_MAPPINGS

epicsarchiver.retrieval.pb.ESC_BYTE = b'\x1b'

epicsarchiver.retrieval.pb.NL_BYTE = b'\n'

epicsarchiver.retrieval.pb.CR_BYTE = b'\r'

epicsarchiver.retrieval.pb.PB_REPLACEMENTS_ESCAPING

epicsarchiver.retrieval.pb.PB_REPLACEMENTS_UNESCAPING

epicsarchiver.retrieval.pb.EeScalarEvent

epicsarchiver.retrieval.pb.EeVectorEvent

epicsarchiver.retrieval.pb.EeEvent

epicsarchiver.retrieval.pb.unescape_bytes(byte_seq: bytes) → bytes

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

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_pd_timestamp(year: int, event: EeEvent) → pandas.Timestamp

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.event_timestamp(year: int, event: EeEvent) → datetime.datetime

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]

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]]]

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) → epicsarchiver.retrieval.archive_event.ArchiveEvent

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

Returns:

ArchiveEvent

epicsarchiver.retrieval.pb.parse_pb_data(raw_data: bytes) → list[epicsarchiver.retrieval.archive_event.ArchiveEvent]

Turn raw PB data into an ArchiveData object.

Parameters:

raw_data – The raw data

Returns:

An ArchiveData object

epicsarchiver.retrieval.pb.get_iso_timestamp_for_event(year: int, event: EeEvent) → str

Returns an ISO-formatted timestamp string for the given event.

epicsarchiver.retrieval.pb.read_pb_file(filename: str) → list[epicsarchiver.retrieval.archive_event.ArchiveEvent]

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

From the protobuf variant.

Parameters:

f (ee.FieldValue) – protobuf field value

Returns:

Basic Field Value

Return type:

FieldValue