epicsarchiver_mgmt.archiver =========================== .. py:module:: epicsarchiver_mgmt.archiver .. autoapi-nested-parse:: Archiver Appliance Management sub package. Provides methods and classes for managing the Archiver Appliance. Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/epicsarchiver_mgmt/archiver/base/index /autoapi/epicsarchiver_mgmt/archiver/info/index /autoapi/epicsarchiver_mgmt/archiver/mgmt/index Exceptions ---------- .. autoapisummary:: epicsarchiver_mgmt.archiver.ArchiverConnectionError epicsarchiver_mgmt.archiver.ArchiverError epicsarchiver_mgmt.archiver.ArchiverResponseError Classes ------- .. autoapisummary:: epicsarchiver_mgmt.archiver.ArchiverMgmtInfo epicsarchiver_mgmt.archiver.ArchivingStatus epicsarchiver_mgmt.archiver.ArchiverMgmt Package Contents ---------------- .. py:exception:: ArchiverConnectionError(base_url: str, message: str | None = None) Bases: :py:obj:`ArchiverError` Exception raised when there is a connection error with the archiver. .. py:attribute:: base_url .. py:exception:: ArchiverError Bases: :py:obj:`epicsarchiver_mgmt.exceptions.BaseMgmtError` Base class for all exceptions raised by the archiver HTTP client. .. py:exception:: ArchiverResponseError(base_url: str, url: str | None = None, response: str | None = None, message: str | None = None) Bases: :py:obj:`ArchiverError` Exception raised when the archiver returns an unexpected response. .. py:attribute:: base_url .. py:attribute:: url :value: None .. py:attribute:: response :value: None .. py:class:: ArchiverMgmtInfo(hostname: str = 'localhost', port: int = 17665) Bases: :py:obj:`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. :param hostname: EPICS Archiver Appliance hostname [default: localhost] :param 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*") .. py:method:: get_all_expanded_pvs() -> list[str] 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 .. py:method:: get_all_pvs(pv_query: str | None = None, regex: str | None = None, limit: int = 500) -> list[str] Return all the PVs in the cluster. :param pv_query: An optional argument that can contain a GLOB wildcard. Will return PVs that match this GLOB. For example: pv=KLYS* :type pv_query: str :param regex: An optional argument that can contain a Java regex wildcard. Will return PVs that match this regex. :type regex: str :param limit: 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. :type limit: int :returns: list of PV names :rtype: list[str] .. py:method:: get_pv_status(pv: str | list[str]) -> InfoResultList Return the status of a PV. :param 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 .. py:method:: get_archiving_status(pv: str) -> ArchivingStatus | None Return the status of a PV. :param pv: name of the pv. :returns: string representing the status .. py:method:: get_pv_details(pv: str | list[str]) -> InfoResultList Return the details of a PV. :param 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 .. py:method:: get_unarchived_pvs(pvs: str | list[str]) -> list[str] Return the list of unarchived PVs out of PVs specified in pvs. :param pvs: a list of PVs either in CSV format or as a python string list :returns: list of unarchived PV names .. py:method:: get_archived_pvs(pvs: str | list[str]) -> list[str] Return the list of archived PVs out of PVs specified in pvs. :param pvs: a list of PVs either in CSV format or as a python string list :returns: list of archived PV names .. py:method:: get_pv_type_info(pv: str) -> TypeInfo Return the type info of a PV. :param pv: name of the pv. :returns: dict with the type info of the matching PVs. .. py:class:: ArchivingStatus Bases: :py:obj:`enum.StrEnum` Enum of archiving status in the archiver. .. py:attribute:: Paused :value: 'Paused' .. py:attribute:: BeingArchived :value: 'Being archived' .. py:attribute:: NotBeingArchived :value: 'Not being archived' .. py:method:: from_str(desc: str) -> ArchivingStatus | None :classmethod: Convert from a string to ArchivingStatus. :param desc: input string :type desc: str :returns: An enum representation. :rtype: ArchivingStatus | None .. py:class:: ArchiverMgmt(hostname: str = 'localhost', port: int = 17665) Bases: :py:obj:`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. :param hostname: EPICS Archiver Appliance hostname [default: localhost] :param port: EPICS Archiver Appliance management port [default: 17665] .. py:method:: 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 Archive a PV. :param pv: PV name. :type pv: str :param sampling_period: The sampling period, i.e. 1.0 is 1Hz. Defaults to None. :type sampling_period: str | None, optional :param sampling_method: The sampling method, SCAN or MONITOR. Defaults to None. :type sampling_method: SamplingMethod | None, optional :param controlling_pv: A pv to control when to archive this pv. Defaults to None. :type controlling_pv: str | None, optional :param policy: The policy, can be found at /mgmt/bpl/getPolicyList. Defaults to None. :type policy: str | None, optional :param appliance: Can specify a specific appliance. Defaults to None. :type appliance: str | None, optional :param protocol: Protocol to use. Defaults to EpicsProto.CA. :type protocol: EpicsProto, optional :returns: _description_ :rtype: OperationResultList .. py:method:: archive_pv_requests(pv_requests: list[ArchivePVRequest]) -> OperationResultList Archive a list of PVs with the given parameters. :param pv_requests: The pv requests. :type pv_requests: list[ArchivePVRequest] :returns: Result of the operation. :rtype: OperationResultList .. py:method:: pause_pv(pv: str) -> OperationResult Pause the archiving of a PV(s). :param pv: name of the pv. Can be a GLOB wildcards or a list of comma separated names. :returns: list of submitted PVs .. py:method:: resume_pv(pv: str) -> OperationResult Resume the archiving of a PV(s). :param pv: name of the pv. Can be a GLOB wildcards or a list of comma separated names. :returns: list of submitted PVs .. py:method:: abort_pv(pv: str) -> OperationResult Abort any pending requests for archiving this PV. :param pv: name of the pv. :returns: Status of action and description. Example: {"status":"ok","desc":"Aborted request for archiving PV PV1"} :rtype: OperationResult .. py:method:: add_alias(pv: str, alias_name: str) -> OperationResult Add an alias to a pv. :param pv: PV to add alias. :param alias_name: name of alias to add to pv. :returns: Status of action and description. :rtype: OperationResult .. py:method:: remove_alias(pv: str, alias_name: str) -> OperationResult Remove an alias to a pv. :param pv: PV to remove alias. :param alias_name: name of alias to remove from pv. :returns: Status of action and description. :rtype: OperationResult .. py:method:: delete_pv(pv: str, delete_data: bool = False) -> OperationResult Stop archiving the specified PV. The PV needs to be paused first. :param pv: name of the pv. :param delete_data: delete the data that has already been recorded. Default to False. :returns: list of submitted PVs .. py:method:: rename_pv(pv: str, newname: str) -> OperationResult Rename this pv to a new name. The PV needs to be paused first. :param pv: name of the pv. :type pv: str :param newname: new name of the pv :type newname: str :returns: Status of action and description. Example: {"status":"ok","desc":"Successfully renamed PV PV1 to PV2"} :rtype: OperationResult .. py:method:: update_pv(pv: str, samplingperiod: float | None = None, samplingmethod: str | None = None) -> OperationResult Change the archival parameters for a PV. :param pv: name of the pv. :param samplingperiod: the new sampling period in seconds. :param samplingmethod: the new sampling method [SCAN|MONITOR] :returns: list of submitted PV .. py:method:: rename_and_append(old: str, new: str, storage: Storage) -> OperationResult 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. :param old: The name of the older pv. The data for this PV will be appended to the newer PV and then deleted. :type old: str :param new: The name of the newer pv. :type new: str :param storage: The name of the store to consolidate data before appending. :type storage: Storage :returns: Result of the operation :rtype: OperationResultList .. py:method:: change_type(pv: str, new_type: ArchDbrType) -> OperationResult Change the type of a pv to a new type. :param pv: Name of the PV :type pv: str :param new_type: New DBR_TYPE :type new_type: ArchDbrType :returns: OperationResult .. py:method:: put_pv_type_info(pv: str, type_info: epicsarchiver_mgmt.archiver.info.TypeInfo, put_info_type: PutInfoType) -> epicsarchiver_mgmt.archiver.info.TypeInfo Put the type info for a PV. :param pv: Name of the PV :type pv: str :param type_info: Type info :type type_info: InfoResult :param put_info_type: Whether override or create new :type put_info_type: PutInfoType :returns: The updated type info :rtype: OperationResult .. py:method:: get_policy_list() -> dict[str, str] Get the list of policies. :returns: dictionary of policies names and descriptions .. py:property:: never_connected_pvs :type: epicsarchiver_mgmt.archiver.info.InfoResultList Get the never connected PVs of the archiver. :returns: List of never connected items. :rtype: InfoResultList .. py:property:: appliances_in_cluster :type: list[dict[str, str]] Get the appliances in the cluster. :returns: List of appliances in the cluster. :rtype: InfoResultList