archiver_test.get

Getting and munging data from the archiver.

Module Contents

Functions

munge_data(...)	Converts the data straight from the archiver to
event_list_to_dict(→ dict[str, epicsarchiver.ArchiveEvent])	Converts raw archiver data to a dictionary of events with isoformat
get_details(→ dict[archiver_test.details.Detail, str])	Get the PV Details for a given pv and input archiver.
archiver_get_dataset(...)	Retrieve archived data and convert to the internal Dataset format
get_data(→ dict[str, ...)	Retrieve archived data for multiple pvs
archiver_get_version(→ str)	Retrieve archiver version

Attributes

LOG

archiver_test.get.LOG: logging.Logger

async archiver_test.get.munge_data(pv: str, archiver_data: list[epicsarchiver.ArchiveEvent] | None, pv_status: dict[archiver_test.details.Detail, str], means_data: list[epicsarchiver.ArchiveEvent] | None) → archiver_test.pv_archive_events.PVArchiveEvents

Converts the data straight from the archiver to an intermediary format that is easier to read.

Parameters:

• archiver_data – list of archiver events

• pv – name of pv

• pv_status – Details of pv from getPVDetails

• means_data – Same as archiver_data but expected to be calculated means value events.

archiver_test.get.event_list_to_dict(a_data: list[epicsarchiver.ArchiveEvent] | None) → dict[str, epicsarchiver.ArchiveEvent]

Converts raw archiver data to a dictionary of events with isoformat

datetimestamp keys.

Parameters:

data – The raw archiver json data.

async archiver_test.get.get_details(archiver: epicsarchiver.ArchiverAppliance, pv: str) → dict[archiver_test.details.Detail, str]

Get the PV Details for a given pv and input archiver.

Parameters:

• archiver – The archiver to query

• pv – name of pv

async archiver_test.get.archiver_get_dataset(pv: str, sop: datetime.datetime, eop: datetime.datetime, archiver: epicsarchiver.ArchiverAppliance) → archiver_test.pv_archive_events.PVArchiveEvents

Retrieve archived data and convert to the internal Dataset format

Parameters:

• archiver – The archiver to query

• pv – name of the pv.

• sop – start time. Can be a string or datetime.datetime object.

• eop – end time. Can be a string or datetime.datetime object.

Returns:

Dataset

async archiver_test.get.get_data(pvs: list[str], sop: datetime.datetime, eop: datetime.datetime, archiver_url: str) → dict[str, archiver_test.pv_archive_events.PVArchiveEvents]

Retrieve archived data for multiple pvs

Parameters:

• archiver_url – The archiver hostname to query

• pvs – name of the pvs to check.

• sop – start time. Can be a string or datetime.datetime object.

• eop – end time. Can be a string or datetime.datetime object.

Returns:

dictionary of Datasets with pv names as keys.

async archiver_test.get.archiver_get_version(archiver_url: str) → str

Retrieve archiver version

Parameters:

archiver_url – The archiver hostname to query

Returns:

version name from archiver