Skip to content

state_analysis

frequenz.lib.notebooks.reporting.state_analysis ¤

Functions for analyzing microgrid component state transitions and extracting alerts.

Classes¤

Functions¤

frequenz.lib.notebooks.reporting.state_analysis.fetch_and_extract_state_durations async ¤

fetch_and_extract_state_durations(
    *,
    client: ReportingApiClient,
    microgrid_components: list[tuple[int, list[int]]],
    metrics: list[Metric],
    start_time: datetime,
    end_time: datetime,
    resampling_period: timedelta | None,
    alert_states: list[ComponentStateCode],
    include_warnings: bool = True
) -> tuple[list[StateRecord], list[StateRecord]]

Fetch data using the Reporting API and extract state durations and alert records.

PARAMETER DESCRIPTION
client

The client used to fetch the metric samples from the Reporting API.

TYPE: ReportingApiClient

microgrid_components

List of tuples where each tuple contains microgrid ID and corresponding component IDs.

TYPE: list[tuple[int, list[int]]]

metrics

List of metric names. NOTE: The service will support requesting states without metrics in the future and this argument will be removed.

TYPE: list[Metric]

start_time

The start date and time for the period.

TYPE: datetime

end_time

The end date and time for the period.

TYPE: datetime

resampling_period

The period for resampling the data. If None, data will be returned in its original resolution.

TYPE: timedelta | None

alert_states

List of ComponentStateCode names that should trigger an alert.

TYPE: list[ComponentStateCode]

include_warnings

Whether to include warning states in the alert records.

TYPE: bool DEFAULT: True

RETURNS DESCRIPTION
tuple[list[StateRecord], list[StateRecord]]

A tuple containing: - A list of StateRecord instances representing the state changes. - A list of StateRecord instances that match the alert criteria.
Source code in frequenz/lib/notebooks/reporting/state_analysis.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
async def fetch_and_extract_state_durations(
    *,
    client: ReportingApiClient,
    microgrid_components: list[tuple[int, list[int]]],
    metrics: list[Metric],
    start_time: datetime,
    end_time: datetime,
    resampling_period: timedelta | None,
    alert_states: list[ComponentStateCode],
    include_warnings: bool = True,
) -> tuple[list[StateRecord], list[StateRecord]]:
    """Fetch data using the Reporting API and extract state durations and alert records.

    Args:
        client: The client used to fetch the metric samples from the Reporting API.
        microgrid_components: List of tuples where each tuple contains microgrid
            ID and corresponding component IDs.
        metrics: List of metric names.
            NOTE: The service will support requesting states without metrics in
            the future and this argument will be removed.
        start_time: The start date and time for the period.
        end_time: The end date and time for the period.
        resampling_period: The period for resampling the data. If None, data
            will be returned in its original resolution.
        alert_states: List of ComponentStateCode names that should trigger an alert.
        include_warnings: Whether to include warning states in the alert records.

    Returns:
        A tuple containing:
            - A list of StateRecord instances representing the state changes.
            - A list of StateRecord instances that match the alert criteria.
    """
    samples = await _fetch_component_data(
        client=client,
        microgrid_components=microgrid_components,
        metrics=metrics,
        start_time=start_time,
        end_time=end_time,
        resampling_period=resampling_period,
        include_states=True,
        include_bounds=False,
    )

    all_states = _extract_state_records(samples, include_warnings)
    alert_records = _filter_alerts(all_states, alert_states, include_warnings)
    return all_states, alert_records