epicsarchiver.statistics.reports.archiver_report

Generate a report detailing a list of statistics of Archiver pvs.

Examples

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

LOG

Classes

ArchiverReport	Configuration for generating the report.
_EnhancedJSONEncoder	Extensible JSON <https://json.org> encoder for Python data structures.

Functions

_is_greater_than_time_minimum(→ bool)
_organise_by_ioc(...)
_iocs_summary(...)
csv_output(→ list[list[str]])	Creates a list[str] output of the generated data for printing as csv.
_invert_data(→ dict[str, ...)	Inverts data from being by statistic, to be by PV.

Module Contents

epicsarchiver.statistics.reports.archiver_report.LOG: logging.Logger

epicsarchiver.statistics.reports.archiver_report._is_greater_than_time_minimum(in_time: datetime.datetime | None, time_minimum: datetime.timedelta) → bool

class epicsarchiver.statistics.reports.archiver_report.ArchiverReport

Configuration for generating the report.

query_limit: int | None

time_minimum: datetime.timedelta

connection_drops_minimum: int

config_options: epicsarchiver.statistics.configuration.ConfigOptions | None

other_archiver: epicsarchiver.statistics.services.archiver_statistics.ArchiverWrapper | None

mb_per_day_minimum: float

events_dropped_minimum: int

channelfinder: epicsarchiver.statistics.services.channelfinder.ChannelFinder

ioc_name: str | None

async _get_responses(statistic: epicsarchiver.statistics.models.stats.Stat, archiver: epicsarchiver.statistics.services.archiver_statistics.ArchiverWrapper) → collections.abc.Sequence[epicsarchiver.statistics.models.stat_responses.BaseStatResponse]

Produce a list of PVs and stats.

Parameters:

• statistic (Stat) – Statistic to fetch

• archiver (ArchiverWrapper) – Archiver to request against

Returns:

Sequence of statistic responses

Return type:

Sequence[BaseStatResponse]

async generate_stats(statistic: epicsarchiver.statistics.models.stats.Stat, archiver: epicsarchiver.statistics.services.archiver_statistics.ArchiverWrapper) → dict[str, epicsarchiver.statistics.models.stat_responses.BaseStatResponse]

Produce a list of PVs and stats.

Parameters:

• statistic (Stat) – Statistic to generate data from

• archiver (ArchiverWrapper) – Archiver to check against

Returns:

dictionary of pvs to statistics

Return type:

dict[str, BaseStatResponse]

async generate(archiver: epicsarchiver.statistics.services.archiver_statistics.ArchiverWrapper) → dict[epicsarchiver.statistics.models.stat_responses.Ioc, dict[str, epicsarchiver.statistics.models.stats.PVStats]]

Generate all the statistics available from the Stat list.

Parameters:

archiver (ArchiverWrapper) – Archiver to get statistics from

Returns:

Return a dictionary with pv names as keys, and detailed statistics after.

Return type:

dict[Ioc, dict[str, PVStats]]

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.

Parameters:

• archiver (ArchiverWrapper) – Archiver to get statistics

• file (IO[str]) – file to print the report to

• verbose (bool, optional) – Verbose output or not. Defaults to False.

class epicsarchiver.statistics.reports.archiver_report._EnhancedJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

Bases: json.JSONEncoder

Extensible JSON <https://json.org> 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).

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)

async epicsarchiver.statistics.reports.archiver_report._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]]

epicsarchiver.statistics.reports.archiver_report._iocs_summary(iocs: dict[epicsarchiver.statistics.models.stat_responses.Ioc, list[str]]) → list[tuple[epicsarchiver.statistics.models.stat_responses.Ioc, int]]

epicsarchiver.statistics.reports.archiver_report.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

Parameters:

report (dict[str, PVStats]) – Base input data in form of pv mapped to Stat and responses from the archiver.

Returns:

List of list of strings

Return type:

list[list[str]]

epicsarchiver.statistics.reports.archiver_report._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.

Parameters:

data (dict[Stat, dict[str, BaseStatResponse]]) – Input data with Stats to dictionary with pv name keys

Returns:

Output with Pv name keys.

Return type:

dict[str, PVStats]