epicsarchiver_mgmt.archiver

Archiver Appliance Management sub package.

Provides methods and classes for managing the Archiver Appliance.

Submodules

Exceptions

`ArchiverConnectionError`	Exception raised when there is a connection error with the archiver.
`ArchiverError`	Base class for all exceptions raised by the archiver HTTP client.
`ArchiverResponseError`	Exception raised when the archiver returns an unexpected response.

Classes

`ArchiverMgmtInfo`	Mgmt Info EPICS Archiver Appliance client.
`ArchivingStatus`	Enum of archiving status in the archiver.
`ArchiverMgmt`	Mgmt Operations EPICS Archiver Appliance client.

Package Contents

exception epicsarchiver_mgmt.archiver.ArchiverConnectionError(base_url: str, message: str | None = None)[source]

Bases: ArchiverError

Exception raised when there is a connection error with the archiver.

base_url

exception epicsarchiver_mgmt.archiver.ArchiverError[source]

Bases: epicsarchiver_mgmt.exceptions.BaseMgmtError

Base class for all exceptions raised by the archiver HTTP client.

exception epicsarchiver_mgmt.archiver.ArchiverResponseError(base_url: str, url: str | None = None, response: str | None = None, message: str | None = None)[source]

Bases: ArchiverError

Exception raised when the archiver returns an unexpected response.

base_url

url = None

response = None

class epicsarchiver_mgmt.archiver.ArchiverMgmtInfo(hostname: str = 'localhost', port: int = 17665)[source]

Bases: epicsarchiver_mgmt.archiver.base.BaseArchiverAppliance

Mgmt Info EPICS Archiver Appliance client.

Hold a session to the Archiver Appliance web application and use the mgmt interface.

Parameters:

hostname – EPICS Archiver Appliance hostname [default: localhost]
port – EPICS Archiver Appliance management port [default: 17665]

Examples: .. code-block:: python

from epicsarchiver_mgmt.archiver.info import ArchiverMgmtInfo

archappl = ArchiverMgmtInfo(“archiver.example.org”) print(archappl.version) archappl.get_pv_status(pv=”BPM*”)

get_all_expanded_pvs() → list[str][source]

Return all expanded PV names in the cluster.

This is targeted at automation and should return the PVs being archived, the fields, .VAL’s, aliases and PV’s in the archive workflow. Note this call can return 10’s of millions of names.

Returns:: list of expanded PV names

get_all_pvs(pv_query: str | None = None, regex: str | None = None, limit: int = 500) → list[str][source]

Return all the PVs in the cluster.

Parameters:

pv_query (str) – An optional argument that can contain a GLOB wildcard. Will return PVs that match this GLOB. For example: pv=KLYS*
regex (str) – An optional argument that can contain a Java regex wildcard. Will return PVs that match this regex.
limit (int) – number of matched PV’s that are returned. To get all the PV names, (potentially in the millions), set limit to -1. Default to 500.

Returns:

list of PV names

Return type:

list[str]

get_pv_status(pv: str | list[str]) → InfoResultList[source]

Return the status of a PV.

Parameters:: pv – name(s) of the pv for which the status is to be determined. Can be a GLOB wildcards or multiple PVs as a comma separated list.
Returns:: list of dict with the status of the matching PVs

get_archiving_status(pv: str) → ArchivingStatus | None[source]

Return the status of a PV.

Parameters:: pv – name of the pv.
Returns:: string representing the status

get_pv_details(pv: str | list[str]) → InfoResultList[source]

Return the details of a PV.

Parameters:: pv – name(s) of the pv for which the details are to be determined. Can be a GLOB wildcards or multiple PVs as a comma separated list.
Returns:: list of dict with the details of the matching PVs

get_unarchived_pvs(pvs: str | list[str]) → list[str][source]

Return the list of unarchived PVs out of PVs specified in pvs.

Parameters:: pvs – a list of PVs either in CSV format or as a python string list
Returns:: list of unarchived PV names

get_archived_pvs(pvs: str | list[str]) → list[str][source]

Return the list of archived PVs out of PVs specified in pvs.

Parameters:: pvs – a list of PVs either in CSV format or as a python string list
Returns:: list of archived PV names

get_pv_type_info(pv: str) → TypeInfo[source]

Return the type info of a PV.

Parameters:: pv – name of the pv.
Returns:: dict with the type info of the matching PVs.

class epicsarchiver_mgmt.archiver.ArchivingStatus[source]

Bases: enum.StrEnum

Enum of archiving status in the archiver.

Paused = 'Paused'

BeingArchived = 'Being archived'

NotBeingArchived = 'Not being archived'

classmethod from_str(desc: str) → ArchivingStatus | None[source]

Convert from a string to ArchivingStatus.

Parameters:: desc (str) – input string
Returns:: An enum representation.
Return type:: ArchivingStatus | None

class epicsarchiver_mgmt.archiver.ArchiverMgmt(hostname: str = 'localhost', port: int = 17665)[source]

Bases: epicsarchiver_mgmt.archiver.info.ArchiverMgmtInfo

Mgmt Operations EPICS Archiver Appliance client.

Hold a session to the Archiver Appliance web application and use the mgmt interface.

Parameters:

hostname – EPICS Archiver Appliance hostname [default: localhost]
port – EPICS Archiver Appliance management port [default: 17665]

archive_pv(pv: str, *, sampling_period: float | None = None, sampling_method: SamplingMethod | None = None, controlling_pv: str | None = None, policy: str | None = None, appliance: str | None = None, protocol: EpicsProto = EpicsProto.CA) → OperationResultList[source]

Archive a PV.

Parameters:

pv (str) – PV name.
sampling_period (str | None, optional) – The sampling period, i.e. 1.0 is 1Hz. Defaults to None.
sampling_method (SamplingMethod | None, optional) – The sampling method, SCAN or MONITOR. Defaults to None.
controlling_pv (str | None, optional) – A pv to control when to archive this pv. Defaults to None.
policy (str | None, optional) – The policy, can be found at /mgmt/bpl/getPolicyList. Defaults to None.
appliance (str | None, optional) – Can specify a specific appliance. Defaults to None.
protocol (EpicsProto, optional) – Protocol to use. Defaults to EpicsProto.CA.

Returns:

_description_

Return type:

OperationResultList

archive_pv_requests(pv_requests: list[ArchivePVRequest]) → OperationResultList[source]

Archive a list of PVs with the given parameters.

Parameters:: pv_requests (list[ArchivePVRequest]) – The pv requests.
Returns:: Result of the operation.
Return type:: OperationResultList

pause_pv(pv: str) → OperationResult[source]

Pause the archiving of a PV(s).

Parameters:: pv – name of the pv. Can be a GLOB wildcards or a list of comma separated names.
Returns:: list of submitted PVs

resume_pv(pv: str) → OperationResult[source]

Resume the archiving of a PV(s).

Parameters:: pv – name of the pv. Can be a GLOB wildcards or a list of comma separated names.
Returns:: list of submitted PVs

abort_pv(pv: str) → OperationResult[source]

Abort any pending requests for archiving this PV.

Parameters:: pv – name of the pv.
Returns:: Status of action and description. Example: {“status”:”ok”,”desc”:”Aborted request for archiving PV PV1”}
Return type:: OperationResult

add_alias(pv: str, alias_name: str) → OperationResult[source]

Add an alias to a pv.

Parameters:

pv – PV to add alias.
alias_name – name of alias to add to pv.

Returns:

Status of action and description.

Return type:

OperationResult

remove_alias(pv: str, alias_name: str) → OperationResult[source]

Remove an alias to a pv.

Parameters:

pv – PV to remove alias.
alias_name – name of alias to remove from pv.

Returns:

Status of action and description.

Return type:

OperationResult

delete_pv(pv: str, delete_data: bool = False) → OperationResult[source]

Stop archiving the specified PV.

The PV needs to be paused first.

Parameters:

pv – name of the pv.
delete_data – delete the data that has already been recorded. Default to False.

Returns:

list of submitted PVs

rename_pv(pv: str, newname: str) → OperationResult[source]

Rename this pv to a new name.

The PV needs to be paused first.

Parameters:

pv (str) – name of the pv.
newname (str) – new name of the pv

Returns:

Status of action and description. Example:: {“status”:”ok”,”desc”:”Successfully renamed PV PV1 to PV2”}

Return type:

OperationResult

update_pv(pv: str, samplingperiod: float | None = None, samplingmethod: str | None = None) → OperationResult[source]

Change the archival parameters for a PV.

Parameters:

pv – name of the pv.
samplingperiod – the new sampling period in seconds.
samplingmethod – the new sampling method [SCAN|MONITOR]

Returns:

list of submitted PV

rename_and_append(old: str, new: str, storage: Storage) → OperationResult[source]

Appends the data for an older PV into a newer PV.

The older PV is deleted and an alias mapping the older PV name to the new PV is added.

Parameters:

old (str) – The name of the older pv. The data for this PV will be appended to the newer PV and then deleted.
new (str) – The name of the newer pv.
storage (Storage) – The name of the store to consolidate data before appending.

Returns: