epicsarchiver.retrieval.command

Command module.

Attributes

LOG

DATE_FORMATS

AlignedPVEvents

Functions

get(→ None)

Print out data from an archiver cluster.

search(→ None)

Search for PV names using regex search patterns.

_meta_field_values(→ dict[int, dict[str, str]])

_table_caption(→ str | None)

_search_table_title(→ str)

_table_title(→ str)

_create_multi_table(→ rich.table.Table)

_to_local_timestamp_str(→ str)

_val_to_str(→ str)

_create_singular_table(→ rich.table.Table)

_create_pv_name_table(→ rich.table.Table)

filtered_event_field_values(→ list[str])

Provide a list of field values for the given event.

_align_events(→ AlignedPVEvents)

Align events from multiple PVs by timestamp.

_multi_fetch_events(→ AlignedPVEvents)

_single_fetch_events(→ tuple[dict[int, ...)

_pv_name_search(→ list[str])

Module Contents

epicsarchiver.retrieval.command.LOG: logging.Logger[source]
epicsarchiver.retrieval.command.DATE_FORMATS = ['%Y-%m-%d', '%Y-%m-%dT%H:%M:%S', '%Y-%m-%d %H:%M:%S', '%Y-%m-%dT%H:%M:%S.%f', '%Y-%m-%d %H:%M:%S.%f'][source]
epicsarchiver.retrieval.command.AlignedPVEvents[source]
epicsarchiver.retrieval.command.get(ctx: click.core.Context, pvs: tuple[str], start: datetime.datetime, end: datetime.datetime, processor_name: str | None, bin_size: int | None, debug: bool) → None[source]

Print out data from an archiver cluster.

ARGUMENT pvs What pvs to get data of.

Example usage:

epicsarchiver --hostname archiver-01.example.com get PV_NAME1 PV_NAME2
epicsarchiver.retrieval.command.search(ctx: click.core.Context, query: str, start: datetime.datetime | None, end: datetime.datetime | None, limit: int, debug: bool) → None[source]

Search for PV names using regex search patterns.

Optionally specify start and/or end times to only return PVs that recorded data in the specified time range.

ARGUMENT query PV name regex search pattern.

Example usage:

epicsarchiver --hostname archiver-01.example.com search                 "PBI-APTM02:Ctrl-ECAT-100:.*Temp1[2-4].*"

epicsarchiver --hostname archiver-01.example.com search                 "PBI-APTM02:.*" -s "2026-01-06 02:50:00"

epicsarchiver --hostname archiver-01.example.com search                 "(?i)mbl-060RFC:.*:tempambient" -s "2026-01-05"  -e "2026-01-06"
epicsarchiver.retrieval.command._meta_field_values(meta: dict[int, epicsarchiver.retrieval.pb.ArchiveEventsMeta]) → dict[int, dict[str, str]][source]
epicsarchiver.retrieval.command._table_caption(field_values: dict[int, dict[str, str]] | None) → str | None[source]
epicsarchiver.retrieval.command._search_table_title(pvs: list[str], start: datetime.datetime | None, end: datetime.datetime | None) → str[source]
epicsarchiver.retrieval.command._table_title(pvs: tuple[str], start: datetime.datetime, end: datetime.datetime, processor: epicsarchiver.retrieval.archiver_retrieval.processor.Processor | None) → str[source]
epicsarchiver.retrieval.command._create_multi_table(pvs: tuple[str], title: str, events: AlignedPVEvents) → rich.table.Table[source]
epicsarchiver.retrieval.command._to_local_timestamp_str(timestamp: pandas.Timestamp) → str[source]
epicsarchiver.retrieval.command._val_to_str(event: epicsarchiver.retrieval.archive_event.ArchiveEvent | None) → str[source]
epicsarchiver.retrieval.command._create_singular_table(pv: str, title: str, caption: str | None, events: AlignedPVEvents) → rich.table.Table[source]
epicsarchiver.retrieval.command._create_pv_name_table(pv_list: list[str], title: str) → rich.table.Table[source]
epicsarchiver.retrieval.command.filtered_event_field_values(fields: list[str], event: epicsarchiver.retrieval.archive_event.ArchiveEvent) → list[str][source]

Provide a list of field values for the given event.

Parameters:
Returns:

Field values for the given event

Return type:

list[str]

epicsarchiver.retrieval.command._align_events(all_events: dict[str, list[epicsarchiver.retrieval.archive_event.ArchiveEvent]]) → AlignedPVEvents[source]

Align events from multiple PVs by timestamp.

Parameters:

all_events (dict[str, list[ArchiveEvent]]) – Events per PV with PV name as key

Returns:

List of pairs, (timestamp, dict[pv_name, pv_value])

Return type:

AlignedPVEvents

async epicsarchiver.retrieval.command._multi_fetch_events(archiver: epicsarchiver.epicsarchiver.ArchiverAppliance, pvs: list[str], start: datetime.datetime, end: datetime.datetime, processor: epicsarchiver.retrieval.archiver_retrieval.processor.Processor | None) → AlignedPVEvents[source]
async epicsarchiver.retrieval.command._single_fetch_events(archiver: epicsarchiver.epicsarchiver.ArchiverAppliance, pv: str, start: datetime.datetime, end: datetime.datetime, processor: epicsarchiver.retrieval.archiver_retrieval.processor.Processor | None) → tuple[dict[int, epicsarchiver.retrieval.pb.ArchiveEventsMeta], AlignedPVEvents][source]