"""Handles the PVDetails report per PV from the Archiver."""
from __future__ import annotations
from enum import Enum
from typing import Dict
from epicsarchiver.statistics.models.stat_responses import (
BaseStatResponse,
ConnectionStatus,
DisconnectedPVsResponse,
DroppedPVResponse,
DroppedReason,
LostConnectionsResponse,
SilentPVsResponse,
StorageRatesResponse,
parse_archiver_datetime,
)
from epicsarchiver.statistics.models.stats import Stat
[docs]
class DetailEnum(str, Enum):
"""Enum of detail types present in the archiver per PV."""
[docs]
DBRType = "Archiver DBR type (from typeinfo):"
[docs]
PVAccess = "Are we using PVAccess?"
[docs]
Method = "Sampling method:"
[docs]
Period = "Sampling period:"
[docs]
Capacity = "Sample buffer capacity"
[docs]
EventRate = "Estimated event rate (events/sec)"
[docs]
EventsLost = "How many events lost totally so far?"
[docs]
TotalEvents = "How many events so far?"
[docs]
Connnected = "Is this PV currently connected?"
[docs]
LastLostConnection = "When did we last lose a connection to this PV?"
[docs]
LostConnections = (
"How many times have we lost and regained the connection to this PV?"
)
[docs]
LostEventsTimestamp = (
"How many events lost because the timestamp is "
"in the far future or past so far?"
)
[docs]
LostEventsBuffer = "How many events lost because the sample buffer is full so far?"
[docs]
LostEventsType = (
"How many events lost because the DBR_Type of the PV "
"has changed from what it used to be?"
)
[docs]
MBStorageRate = "Estimated storage rate (MB/day)"
[docs]
LastEvent = "When did we receive the last event?"
[docs]
Hostname = "Hostname of PV from CA"
[docs]
Instance = "Instance archiving PV"
[docs]
CommandThread = "The CAJ command thread id"
@classmethod
[docs]
def from_str(cls, desc: str) -> DetailEnum | None:
"""Convert from a string to DetailEnum.
Args:
desc (str): input string
Returns:
DetailEnum | None: An enum representation.
"""
for e in DetailEnum:
if e.value == desc:
return e
return None
[docs]
def _dropped_pv_response(
pv_name: str, value: str, reason: DroppedReason, stat: Stat
) -> tuple[Stat, DroppedPVResponse] | None:
if value != "0":
return (
stat,
DroppedPVResponse(pv_name, int(value), reason),
)
return None
[docs]
class Details(Dict[DetailEnum, str]):
"""Representation of the response from the pvDetails endpoint in archiver."""
@classmethod
[docs]
def from_json(cls, json: list[dict[str, str]]) -> Details:
"""Convert from the json representation at the Archiver Endpoint.
Takes some of the details which are relevant for reports to create Details.
Args:
json (list[dict[str, str]]): input json
Returns:
Details: dictionary of DetailEnum to string value.
"""
result: Details = Details()
for json_det in json:
detail_enum = DetailEnum.from_str(json_det["name"])
if detail_enum:
result[detail_enum] = json_det["value"]
return result
[docs]
def to_base_responses(
self, mb_per_day_min: float = 0
) -> dict[Stat, BaseStatResponse]:
"""Convert to a BaseStatResponse dict to match Generic Archiver Statistics.
Args:
mb_per_day_min (float): Minimum MB per day to filter by
Returns:
dict[Stat, BaseStatResponse]: Stat to BaseStatResponse output
"""
result: dict[Stat, BaseStatResponse] = {}
for detail_enum, value in self.items():
response = self.detail_to_base_response(detail_enum, value, mb_per_day_min)
if response:
result[response[0]] = response[1]
return result
[docs]
def detail_to_base_response( # noqa: PLR0911
self, detail_enum: DetailEnum, value: str, mb_per_day_min: float = 0
) -> tuple[Stat, BaseStatResponse] | None:
"""Convert a single detail to a Stat and BaseStatResponse.
Args:
detail_enum (DetailEnum): Detail
value (str): String value of the detail
mb_per_day_min (float): Minimum MB per day to filter by
Returns:
tuple[Stat, BaseStatResponse] | None: Output
"""
pv_name = self[DetailEnum.PVName]
if detail_enum == DetailEnum.LostEventsTimestamp:
return _dropped_pv_response(
pv_name,
value,
DroppedReason.IncorrectTimestamp,
Stat.IncorrectTimestamp,
)
if detail_enum == DetailEnum.LostEventsType:
return _dropped_pv_response(
pv_name, value, DroppedReason.TypeChange, Stat.TypeChange
)
if detail_enum == DetailEnum.LostEventsBuffer:
return _dropped_pv_response(
pv_name, value, DroppedReason.BufferOverflow, Stat.BufferOverflow
)
if detail_enum == DetailEnum.Connnected and value != "yes":
return (
Stat.DisconnectedPVs,
DisconnectedPVsResponse(
pv_name,
self[DetailEnum.Hostname],
parse_archiver_datetime(self[DetailEnum.LastLostConnection]),
self[DetailEnum.Instance],
int(self[DetailEnum.CommandThread]),
0,
parse_archiver_datetime(self[DetailEnum.LastEvent]),
),
)
if detail_enum == DetailEnum.LastEvent and value == "Never":
return (
Stat.SilentPVs,
SilentPVsResponse(pv_name, self[DetailEnum.Instance], None),
)
if detail_enum == DetailEnum.LostConnections and value != "0":
return (
Stat.LostConnection,
LostConnectionsResponse(
pv_name,
ConnectionStatus.CurrentlyConnected
if self[DetailEnum.Connnected] == "yes"
else ConnectionStatus.NotCurrentlyConnected,
self[DetailEnum.Instance],
int(value),
),
)
if detail_enum == DetailEnum.MBStorageRate and (
value != "Not enough info" and float(value) > mb_per_day_min
):
return (
Stat.StorageRates,
StorageRatesResponse(pv_name, float(value), None, None),
)
return None