epicsarchiver.statistics.reports.archiver_report ================================================ .. py:module:: epicsarchiver.statistics.reports.archiver_report .. autoapi-nested-parse:: Generate a report detailing a list of statistics of Archiver pvs. .. rubric:: Examples .. code-block:: python report = ArchiverReport( query_limit=1000, time_minimum=timedelta(days=100), connection_drops_minimum=30, config_options=configuration.ConfigOptions("/config_repo", "tn"), other_archiver=ArchiverAppliance("other_archiver.example.org"), mb_per_day_minimum=1000, events_dropped_minimum=1000, channelfinder=ChannelFinder("channelfinder.tn.ess.lu.se"), ioc_name="AN_IOC_NAME", ) report.print_report(archiver, out_file, verbose=True) Attributes ---------- .. autoapisummary:: epicsarchiver.statistics.reports.archiver_report.LOG Classes ------- .. autoapisummary:: epicsarchiver.statistics.reports.archiver_report.ArchiverReport epicsarchiver.statistics.reports.archiver_report._EnhancedJSONEncoder Functions --------- .. autoapisummary:: epicsarchiver.statistics.reports.archiver_report._is_greater_than_time_minimum epicsarchiver.statistics.reports.archiver_report._organise_by_ioc epicsarchiver.statistics.reports.archiver_report._iocs_summary epicsarchiver.statistics.reports.archiver_report.csv_output epicsarchiver.statistics.reports.archiver_report._invert_data Module Contents --------------- .. py:data:: LOG :type: logging.Logger .. py:function:: _is_greater_than_time_minimum(in_time: datetime.datetime | None, time_minimum: datetime.timedelta) -> bool .. py:class:: ArchiverReport Configuration for generating the report. .. py:attribute:: query_limit :type: int | None .. py:attribute:: time_minimum :type: datetime.timedelta .. py:attribute:: connection_drops_minimum :type: int .. py:attribute:: config_options :type: epicsarchiver.statistics.configuration.ConfigOptions | None .. py:attribute:: other_archiver :type: epicsarchiver.statistics.services.archiver_statistics.ArchiverWrapper | None .. py:attribute:: mb_per_day_minimum :type: float .. py:attribute:: events_dropped_minimum :type: int .. py:attribute:: channelfinder :type: epicsarchiver.statistics.services.channelfinder.ChannelFinder .. py:attribute:: ioc_name :type: str | None .. py:method:: _get_responses(statistic: epicsarchiver.statistics.models.stats.Stat, archiver: epicsarchiver.statistics.services.archiver_statistics.ArchiverWrapper) -> collections.abc.Sequence[epicsarchiver.statistics.models.stat_responses.BaseStatResponse] :async: Produce a list of PVs and stats. :param statistic: Statistic to fetch :type statistic: Stat :param archiver: Archiver to request against :type archiver: ArchiverWrapper :returns: Sequence of statistic responses :rtype: Sequence[BaseStatResponse] .. py:method:: generate_stats(statistic: epicsarchiver.statistics.models.stats.Stat, archiver: epicsarchiver.statistics.services.archiver_statistics.ArchiverWrapper) -> dict[str, epicsarchiver.statistics.models.stat_responses.BaseStatResponse] :async: Produce a list of PVs and stats. :param statistic: Statistic to generate data from :type statistic: Stat :param archiver: Archiver to check against :type archiver: ArchiverWrapper :returns: dictionary of pvs to statistics :rtype: dict[str, BaseStatResponse] .. py:method:: generate(archiver: epicsarchiver.statistics.services.archiver_statistics.ArchiverWrapper) -> dict[epicsarchiver.statistics.models.stat_responses.Ioc, dict[str, epicsarchiver.statistics.models.stats.PVStats]] :async: Generate all the statistics available from the Stat list. :param archiver: Archiver to get statistics from :type archiver: ArchiverWrapper :returns: Return a dictionary with pv names as keys, and detailed statistics after. :rtype: dict[Ioc, dict[str, PVStats]] .. py:method:: print_report(archiver: epicsarchiver.statistics.services.archiver_statistics.ArchiverWrapper, file: IO[str], *, verbose: bool = False) -> None Prints a report about the statistics of PVs in the archiver. :param archiver: Archiver to get statistics :type archiver: ArchiverWrapper :param file: file to print the report to :type file: IO[str] :param verbose: Verbose output or not. Defaults to False. :type verbose: bool, optional .. py:class:: _EnhancedJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None) Bases: :py:obj:`json.JSONEncoder` Extensible JSON encoder for Python data structures. Supports the following objects and types by default: +-------------------+---------------+ | Python | JSON | +===================+===============+ | dict | object | +-------------------+---------------+ | list, tuple | array | +-------------------+---------------+ | str | string | +-------------------+---------------+ | int, float | number | +-------------------+---------------+ | True | true | +-------------------+---------------+ | False | false | +-------------------+---------------+ | None | null | +-------------------+---------------+ To extend this to recognize other objects, subclass and implement a ``.default()`` method with another method that returns a serializable object for ``o`` if possible, otherwise it should call the superclass implementation (to raise ``TypeError``). .. py:method:: default(o: Any) -> Any Implement this method in a subclass such that it returns a serializable object for ``o``, or calls the base implementation (to raise a ``TypeError``). For example, to support arbitrary iterators, you could implement default like this:: def default(self, o): try: iterable = iter(o) except TypeError: pass else: return list(iterable) # Let the base class default method raise the TypeError return super().default(o) .. py:function:: _organise_by_ioc(inverted_report: dict[str, epicsarchiver.statistics.models.stats.PVStats], channelfinder: epicsarchiver.statistics.services.channelfinder.ChannelFinder, ioc_name: str | None = None) -> dict[epicsarchiver.statistics.models.stat_responses.Ioc, dict[str, epicsarchiver.statistics.models.stats.PVStats]] :async: .. py:function:: _iocs_summary(iocs: dict[epicsarchiver.statistics.models.stat_responses.Ioc, list[str]]) -> list[tuple[epicsarchiver.statistics.models.stat_responses.Ioc, int]] .. py:function:: csv_output(report: dict[epicsarchiver.statistics.models.stat_responses.Ioc, dict[str, epicsarchiver.statistics.models.stats.PVStats]]) -> list[list[str]] Creates a list[str] output of the generated data for printing as csv. Outs with headings: IOC Name, IOC hostname, PV name, Statistic, Statistic Note :param report: Base input data in form of pv mapped to Stat and responses from the archiver. :type report: dict[str, PVStats] :returns: List of list of strings :rtype: list[list[str]] .. py:function:: _invert_data(data: dict[epicsarchiver.statistics.models.stats.Stat, dict[str, epicsarchiver.statistics.models.stat_responses.BaseStatResponse]]) -> dict[str, epicsarchiver.statistics.models.stats.PVStats] Inverts data from being by statistic, to be by PV. :param data: Input data with Stats to dictionary with pv name keys :type data: dict[Stat, dict[str, BaseStatResponse]] :returns: Output with Pv name keys. :rtype: dict[str, PVStats]