epicsarchiver.statistics.models.stat_responses ============================================== .. py:module:: epicsarchiver.statistics.models.stat_responses .. autoapi-nested-parse:: Data structures for the statistics endpoints from the archiver. Attributes ---------- .. autoapisummary:: epicsarchiver.statistics.models.stat_responses._DATE_FORMAT_OFFSET epicsarchiver.statistics.models.stat_responses._DATE_FORMAT_TIMEZONE epicsarchiver.statistics.models.stat_responses.UNKNOWN_IOC Classes ------- .. autoapisummary:: epicsarchiver.statistics.models.stat_responses.DroppedReason epicsarchiver.statistics.models.stat_responses.BaseStatResponse epicsarchiver.statistics.models.stat_responses.DroppedPVResponse epicsarchiver.statistics.models.stat_responses.DisconnectedPVsResponse epicsarchiver.statistics.models.stat_responses.SilentPVsResponse epicsarchiver.statistics.models.stat_responses.ConnectionStatus epicsarchiver.statistics.models.stat_responses.LostConnectionsResponse epicsarchiver.statistics.models.stat_responses.StorageRatesResponse epicsarchiver.statistics.models.stat_responses.BothArchiversResponse epicsarchiver.statistics.models.stat_responses.PausedPVResponse epicsarchiver.statistics.models.stat_responses.ConfiguredStatus epicsarchiver.statistics.models.stat_responses.NoConfigResponse epicsarchiver.statistics.models.stat_responses.Ioc epicsarchiver.statistics.models.stat_responses.NameCheckResponse Functions --------- .. autoapisummary:: epicsarchiver.statistics.models.stat_responses.parse_archiver_datetime Module Contents --------------- .. py:class:: DroppedReason Bases: :py:obj:`str`, :py:obj:`enum.Enum` List of reasons why a PV could be dropping events. Includes the endpoints in the archiver corresponding to the reason. .. py:attribute:: IncorrectTimestamp :value: '/getPVsByDroppedEventsTimestamp' .. py:attribute:: BufferOverflow :value: '/getPVsByDroppedEventsBuffer' .. py:attribute:: TypeChange :value: '/getPVsByDroppedEventsTypeChange' .. py:attribute:: SlowChanging :value: '/getSlowChangingPVsWithDroppedEvents' .. py:class:: BaseStatResponse Base class for responses from the archiver statistical endpoints. .. py:attribute:: pv_name :type: str .. py:class:: DroppedPVResponse Bases: :py:obj:`BaseStatResponse` Response from the endpoints in DroppedReason. .. py:attribute:: events_dropped :type: int .. py:attribute:: dropped_reason :type: DroppedReason .. py:method:: from_json(json: dict[str, str], dropped_reason: DroppedReason) -> DroppedPVResponse :classmethod: Convert to DroppedPVResponse from dictionary generated from json. :param json: Input json :type json: dict[str, str] :param dropped_reason: Input reason for events being dropped :type dropped_reason: DroppedReason :returns: The corresponding DroppedPVResponse :rtype: DroppedPVResponse .. py:method:: __str__() -> str Generate a display string for the response. :returns: "Dropped {events} by {reason}" :rtype: str .. py:data:: _DATE_FORMAT_OFFSET :value: '%b/%d/%Y %H:%M:%S %z' .. py:data:: _DATE_FORMAT_TIMEZONE :value: '%b/%d/%Y %H:%M:%S %Z' .. py:function:: parse_archiver_datetime(datetime_str: str) -> datetime.datetime | None Calculate a datetime.datetime from the possible input strings of the archiver. :param datetime_str: Either "%b/%d/%Y %H:%M:%S %z", "Never", "" or "%b/%d/%Y %H:%M:%S %Z" :type datetime_str: str :returns: Datetime representation :rtype: datetime.datetime | None .. py:class:: DisconnectedPVsResponse Bases: :py:obj:`BaseStatResponse` Response from getCurrentlyDisconnectedPVs. Example: .. code-block:: json { "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" } .. py:attribute:: host_name :type: str .. py:attribute:: connection_lost_at :type: datetime.datetime | None .. py:attribute:: instance :type: str .. py:attribute:: command_thread_id :type: int .. py:attribute:: no_connection_as_of_epoch :type: int .. py:attribute:: last_known_event :type: datetime.datetime | None .. py:method:: from_json(json: dict[str, str]) -> DisconnectedPVsResponse :classmethod: Response from the endpoint in getCurrentlyDisconnectedPVs. :param json: Input json :type json: dict[str, str] :returns: Output dataclass :rtype: DisconnectedPVsResponse .. py:method:: __str__() -> str Generate a display string for the response. :returns: "Disconnected {time_difference} ago. Last event at {last_known_event}" :rtype: str .. py:class:: SilentPVsResponse Bases: :py:obj:`BaseStatResponse` Return a list of PVs sorted by the timestamp of the last event received. Example: .. code-block:: json {"pvName":"DTL-030:SC-IOC-002:CA_CLNT_CNT","instance":"archiver-linac-01","lastKnownEvent":"Never"} .. py:attribute:: instance :type: str .. py:attribute:: last_known_event :type: datetime.datetime | None .. py:method:: from_json(json: dict[str, str]) -> SilentPVsResponse :classmethod: Response from the endpoint in getSilentPVsReport. :param json: Input json :type json: dict[str, str] :returns: Output dataclass :rtype: SilentPVsResponse .. py:method:: __str__() -> str Generate a display string for the response. :returns: "No events stored. Last invalid event recieved at {last_known_event}" :rtype: str .. py:class:: ConnectionStatus(*args, **kwds) Bases: :py:obj:`enum.Enum` Connection status enum. :param enum: Placement of enum. :type enum: int .. py:attribute:: CurrentlyConnected .. py:attribute:: NotCurrentlyConnected .. py:class:: LostConnectionsResponse Bases: :py:obj:`BaseStatResponse` Return a list of PVs sorted by the no. of connection drops. Example: .. code-block:: json { "currentlyConnected": "Yes", "pvName": "MBL-010LWU:Vac-VPN-10000:IonCurR", "instance": "archiver-linac-01", "lostConnections": "2586" } .. py:attribute:: currently_connected :type: ConnectionStatus .. py:attribute:: instance :type: str .. py:attribute:: lost_connections :type: int .. py:method:: from_json(json: dict[str, str]) -> LostConnectionsResponse :classmethod: Response from the endpoint in getLostConnectionsReport. :param json: Input json :type json: dict[str, str] :returns: Output dataclass :rtype: LostConnectionsResponse .. py:method:: __str__() -> str Generate a display string for the response. :returns: "Lost connections: {lost_connections}. Connected: {connected}" :rtype: str .. py:class:: StorageRatesResponse Bases: :py:obj:`BaseStatResponse` Return a list of PVs sorted by the no. of connection drops. Example: .. code-block:: json { "pvName": "TS2-010CRM:EMR-XRS-001:mca1", "storageRate_MBperDay": "1099.2894956029622", "storageRate_KBperHour": "46903.01847905972", "storageRate_GBperYear": "391.8365877881653" } .. py:attribute:: mb_per_day :type: float .. py:attribute:: kb_per_hour :type: float | None .. py:attribute:: gb_per_year :type: float | None .. py:method:: from_json(json: dict[str, str]) -> StorageRatesResponse :classmethod: Response from the endpoint in getStorageRateReport. :param json: _description_ :type json: dict[str, str] :returns: _description_ :rtype: StorageRatesResponse .. py:method:: __str__() -> str Generate a display string for the response. :returns: "Storing {self.mb_per_day} MB per day." :rtype: str .. py:class:: BothArchiversResponse Bases: :py:obj:`BaseStatResponse` Response of pvs archived in two archivers. .. py:attribute:: hostname :type: str .. py:attribute:: other_hostname :type: str .. py:method:: __str__() -> str Generate a display string for the response. :returns: "In both {hostname} and {other_hostname}" :rtype: str .. py:class:: PausedPVResponse Bases: :py:obj:`BaseStatResponse` Response of pvs paused. .. py:attribute:: instance :type: str .. py:attribute:: modification_time :type: str .. py:method:: from_json(json: dict[str, str]) -> PausedPVResponse :classmethod: Response from the endpoint in getPausedPVsReport. :param json: Input json format :type json: dict[str, str] :returns: Output dataclass :rtype: PausedPVResponse .. py:method:: __str__() -> str Generate a display string for the response. :returns: "{pv} is paused" :rtype: str .. py:class:: ConfiguredStatus Bases: :py:obj:`str`, :py:obj:`enum.Enum` Represents if a pv is configured in gitlab, channelfinder or archived only. .. py:attribute:: Archived :value: 'Archived but not in config.' .. py:attribute:: ConfiguredGitlab :value: 'Configured in Gitlab but not in archiver.' .. py:attribute:: ConfiguredChannelFinder :value: 'Configured in ChannelFinder but not in archiver.' .. py:class:: NoConfigResponse Bases: :py:obj:`BaseStatResponse` Response of pvs archived but not in configuration files. .. py:attribute:: configured_status :type: ConfiguredStatus .. py:attribute:: alias :type: list[str] .. py:attribute:: alias_archived :type: list[str] .. py:method:: __str__() -> str Generate a display string for the response. :returns: f"Archived but not in config." :rtype: str .. py:method:: __hash__() -> int Hash method for NoConfigResponse. :returns: returns the has of the string representation. :rtype: int .. py:class:: Ioc Minimal info on an Ioc. .. py:attribute:: hostname :type: str .. py:attribute:: name :type: str .. py:method:: from_channel(channel: epicsarchiver.statistics.services.channelfinder.Channel) -> Ioc :classmethod: Gets IOC info from a channel. :param channel: Input channel :type channel: Channel :returns: The Ioc matched to the channel :rtype: Ioc .. py:data:: UNKNOWN_IOC :type: Ioc .. py:class:: NameCheckResponse Bases: :py:obj:`BaseStatResponse` Response of pvs paused. .. py:attribute:: suffix :type: str | None .. py:attribute:: internal :type: bool .. py:method:: __str__() -> str Generate a display string for the response. :returns: "{pv} has a name discouraged to be archived, as {it ends with {suffix}} {it contains #}" :rtype: str