epicsarchiver.statistics.models.stat_responses

Data structures for the statistics endpoints from the archiver.

Attributes

`_DATE_FORMAT_OFFSET`
`_DATE_FORMAT_TIMEZONE`
`UNKNOWN_IOC`

Classes

`DroppedReason`	List of reasons why a PV could be dropping events.
`BaseStatResponse`	Base class for responses from the archiver statistical endpoints.
`DroppedPVResponse`	Response from the endpoints in DroppedReason.
`DisconnectedPVsResponse`	Response from getCurrentlyDisconnectedPVs.
`SilentPVsResponse`	Return a list of PVs sorted by the timestamp of the last event received.
`ConnectionStatus`	Connection status enum.
`LostConnectionsResponse`	Return a list of PVs sorted by the no. of connection drops.
`StorageRatesResponse`	Return a list of PVs sorted by the no. of connection drops.
`BothArchiversResponse`	Response of pvs archived in two archivers.
`PausedPVResponse`	Response of pvs paused.
`ConfiguredStatus`	Represents if a pv is configured in gitlab, channelfinder or archived only.
`NoConfigResponse`	Response of pvs archived but not in configuration files.
`Ioc`	Minimal info on an Ioc.
`NameCheckResponse`	Response of pvs paused.

Functions

parse_archiver_datetime(→ datetime.datetime | None)

Calculate a datetime.datetime from the possible input strings of the archiver.

Module Contents

class epicsarchiver.statistics.models.stat_responses.DroppedReason[source]

Bases: str, enum.Enum

List of reasons why a PV could be dropping events.

Includes the endpoints in the archiver corresponding to the reason.

IncorrectTimestamp = '/getPVsByDroppedEventsTimestamp'[source]

BufferOverflow = '/getPVsByDroppedEventsBuffer'[source]

TypeChange = '/getPVsByDroppedEventsTypeChange'[source]

SlowChanging = '/getSlowChangingPVsWithDroppedEvents'[source]

class epicsarchiver.statistics.models.stat_responses.BaseStatResponse[source]

Base class for responses from the archiver statistical endpoints.

pv_name: str[source]

class epicsarchiver.statistics.models.stat_responses.DroppedPVResponse[source]

Bases: BaseStatResponse

Response from the endpoints in DroppedReason.

events_dropped: int[source]

dropped_reason: DroppedReason[source]

classmethod from_json(json: dict[str, str], dropped_reason: DroppedReason) → DroppedPVResponse[source]

Convert to DroppedPVResponse from dictionary generated from json.

Parameters:

json (dict[str, str]) – Input json
dropped_reason (DroppedReason) – Input reason for events being dropped

Returns:

The corresponding DroppedPVResponse

Return type:

DroppedPVResponse

__str__() → str[source]

Generate a display string for the response.

Returns:: “Dropped {events} by {reason}”
Return type:: str

epicsarchiver.statistics.models.stat_responses._DATE_FORMAT_OFFSET = '%b/%d/%Y %H:%M:%S %z'[source]

epicsarchiver.statistics.models.stat_responses._DATE_FORMAT_TIMEZONE = '%b/%d/%Y %H:%M:%S %Z'[source]

epicsarchiver.statistics.models.stat_responses.parse_archiver_datetime(datetime_str: str) → datetime.datetime | None[source]

Calculate a datetime.datetime from the possible input strings of the archiver.

Parameters:: datetime_str (str) – Either “%b/%d/%Y %H:%M:%S %z”, “Never”, “” or “%b/%d/%Y %H:%M:%S %Z”
Returns:: Datetime representation
Return type:: datetime.datetime | None

class epicsarchiver.statistics.models.stat_responses.DisconnectedPVsResponse[source]

Bases: BaseStatResponse

Response from getCurrentlyDisconnectedPVs.

Example:

{
    "hostName": "N/A",
    "connectionLostAt": "Sep/14/2023 16:00:18 +02:00",
    "pvName": "HCB-ACH:ODH-O2iM-1:O2Level",
    "instance": "sw-vm-11",
    "commandThreadID": "6",
    "noConnectionAsOfEpochSecs": "1694700018",
    "lastKnownEvent": "Aug/25/2023 15:38:17 +02:00"
}

host_name: str[source]

connection_lost_at: datetime.datetime | None[source]

instance: str[source]

command_thread_id: int[source]

no_connection_as_of_epoch: int[source]

last_known_event: datetime.datetime | None[source]

classmethod from_json(json: dict[str, str]) → DisconnectedPVsResponse[source]

Response from the endpoint in getCurrentlyDisconnectedPVs.

Parameters:: json (dict[str, str]) – Input json
Returns:: Output dataclass
Return type:: DisconnectedPVsResponse

__str__() → str[source]

Generate a display string for the response.

Returns:: “Disconnected {time_difference} ago. Last event at {last_known_event}”
Return type:: str

class epicsarchiver.statistics.models.stat_responses.SilentPVsResponse[source]

Bases: BaseStatResponse

Return a list of PVs sorted by the timestamp of the last event received.

Example:

{"pvName":"DTL-030:SC-IOC-002:CA_CLNT_CNT","instance":"archiver-linac-01","lastKnownEvent":"Never"}

instance: str[source]

last_known_event: datetime.datetime | None[source]

classmethod from_json(json: dict[str, str]) → SilentPVsResponse[source]

Response from the endpoint in getSilentPVsReport.

Parameters:: json (dict[str, str]) – Input json
Returns:: Output dataclass
Return type:: SilentPVsResponse

__str__() → str[source]

Generate a display string for the response.

Returns:: “No events stored. Last invalid event recieved at {last_known_event}”
Return type:: str

class epicsarchiver.statistics.models.stat_responses.ConnectionStatus(*args, **kwds)[source]

Bases: enum.Enum

Connection status enum.

Parameters:: enum (int) – Placement of enum.

CurrentlyConnected[source]

NotCurrentlyConnected[source]

class epicsarchiver.statistics.models.stat_responses.LostConnectionsResponse[source]

Bases: BaseStatResponse

Return a list of PVs sorted by the no. of connection drops.

Example:

{
    "currentlyConnected": "Yes",
    "pvName": "MBL-010LWU:Vac-VPN-10000:IonCurR",
    "instance": "archiver-linac-01",
    "lostConnections": "2586"
}

currently_connected: ConnectionStatus[source]

instance: str[source]

lost_connections: int[source]

classmethod from_json(json: dict[str, str]) → LostConnectionsResponse[source]

Response from the endpoint in getLostConnectionsReport.

Parameters:: json (dict[str, str]) – Input json
Returns:: Output dataclass
Return type:: LostConnectionsResponse

__str__() → str[source]

Generate a display string for the response.

Returns:: “Lost connections: {lost_connections}. Connected: {connected}”
Return type:: str

class epicsarchiver.statistics.models.stat_responses.StorageRatesResponse[source]

Bases: BaseStatResponse

Return a list of PVs sorted by the no. of connection drops.

Example:

{
    "pvName": "TS2-010CRM:EMR-XRS-001:mca1",
    "storageRate_MBperDay": "1099.2894956029622",
    "storageRate_KBperHour": "46903.01847905972",
    "storageRate_GBperYear": "391.8365877881653"
}