Skip to content

Index

frequenz.gridpool ¤

High-level interface to grid pools for the Frequenz platform.

Classes¤

frequenz.gridpool.ComponentGraphConfig ¤

Configuration for the component graph.

Source code in frequenz/microgrid_component_graph/__init__.py 
17
18
19
20
    "FormulaGenerationError",
    "FormulaOverrides",
    "InvalidGraphError",
]
Methods:¤
__init__ ¤
__init__(
    *,
    allow_component_validation_failures: bool = False,
    allow_unconnected_components: bool = False,
    allow_unspecified_inverters: bool = False,
    disable_fallback_components: bool = False,
    include_phantom_loads_in_consumer_formula: bool = False,
    prefer_meters_in_component_formulas: bool = False,
    formula_overrides: FormulaOverrides | None = None
) -> None

Initialize this instance.

PARAMETER DESCRIPTION
allow_component_validation_failures

Whether to allow validation errors on components. When this is True, the graph will be built even if there are validation errors on the components.

TYPE: bool DEFAULT: False

allow_unconnected_components

Whether to allow unconnected components in the graph, that are not reachable from the root.

TYPE: bool DEFAULT: False

allow_unspecified_inverters

Whether to allow untyped inverters in the graph. When this is True, inverters that have InverterType::Unspecified will be assumed to be Battery inverters.

TYPE: bool DEFAULT: False

disable_fallback_components

Whether to disable fallback components in generated formulas. When this is True, the formulas will not include fallback components.

TYPE: bool DEFAULT: False

include_phantom_loads_in_consumer_formula

Whether to consider phantom loads in the consumer formula. Meters with successors can still have loads not represented in the component graph. These are called phantom loads. When this is true, phantom loads are included in formulas by excluding the measurements of successor meters from the measurements of their predecessor meters. When false, consumer formula is generated by excluding production and battery components from the grid measurements. The non-consumer components behind one internal meter are excluded as one group: the meter reading first, the component readings as the fallback. While the meter reading is used, a load behind that meter that is not in the component graph is then also excluded together with the group.

TYPE: bool DEFAULT: False

prefer_meters_in_component_formulas

Default policy for the per-category formulas. When False (the default), the component measurement is the primary source and the meter measurement is the fallback for battery_formula, chp_formula, pv_formula, wind_turbine_formula, ev_charger_formula, and steam_boiler_formula. When True, the meter is primary and the component is the fallback. Has no effect on grid_formula, consumer_formula, producer_formula, or any of the coalesce formulas.

TYPE: bool DEFAULT: False

formula_overrides

Per-formula overrides for the meter/component preference; see FormulaOverrides. Each entry, when set, takes precedence over prefer_meters_in_component_formulas for that formula.

TYPE: FormulaOverrides | None DEFAULT: None

Source code in frequenz/microgrid_component_graph/__init__.py 
20
]

frequenz.gridpool.ComponentGraphGenerator ¤

Generates component graphs for microgrids using the Assets API.

Source code in src/frequenz/gridpool/_graph_generator.py 
 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
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class ComponentGraphGenerator:
    """Generates component graphs for microgrids using the Assets API."""

    def __init__(
        self,
        client: AssetsApiClient,
        config: ComponentGraphConfig | None = None,
    ) -> None:
        """Initialize this instance.

        Args:
            client: The Assets API client to use for fetching components and
                connections.
            config: How to build the graph and generate its formulas. See
                `ComponentGraphConfig`. Defaults to that class's own defaults.
        """
        self._client: AssetsApiClient = client
        self._config: ComponentGraphConfig = (
            config if config is not None else ComponentGraphConfig()
        )

    async def get_component_graph(
        self, microgrid_id: MicrogridId
    ) -> MicrogridComponentGraph:
        """Generate a component graph for the given microgrid ID.

        Args:
            microgrid_id: The ID of the microgrid to generate the graph for.

        Returns:
            The component graph representing the microgrid's electrical
                components and their connections.

        Raises:
            ValueError: If any component connections could not be loaded.
        """
        components = await self._client.list_microgrid_electrical_components(
            microgrid_id
        )
        connections = (
            await self._client.list_microgrid_electrical_component_connections(
                microgrid_id
            )
        )

        if any(c is None for c in connections):
            raise ValueError("Failed to load all electrical component connections.")

        breakers = [c for c in components if isinstance(c, Breaker)]
        connected_breakers = [
            b
            for b in breakers
            if any(
                b.id in (c.source, c.destination) for c in connections if c is not None
            )
        ]

        if connected_breakers:
            _logger.warning(
                "The following breakers are connected to other components, "
                + "which is not supported by the component graph generator and may "
                + "lead to graph traversal issues: %s",
                [b.id for b in connected_breakers],
            )
        elif breakers:
            _logger.debug("Dropping unconnected breakers: %s", [b.id for b in breakers])
            components = [c for c in components if not isinstance(c, Breaker)]

        graph = ComponentGraph[
            ElectricalComponent, ComponentConnection, ElectricalComponentId
        ](components, connections, self._config)

        return graph
Methods:¤
__init__ ¤
__init__(
    client: AssetsApiClient,
    config: ComponentGraphConfig | None = None,
) -> None

Initialize this instance.

PARAMETER DESCRIPTION
client

The Assets API client to use for fetching components and connections.

TYPE: AssetsApiClient

config

How to build the graph and generate its formulas. See ComponentGraphConfig. Defaults to that class's own defaults.

TYPE: ComponentGraphConfig | None DEFAULT: None

Source code in src/frequenz/gridpool/_graph_generator.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def __init__(
    self,
    client: AssetsApiClient,
    config: ComponentGraphConfig | None = None,
) -> None:
    """Initialize this instance.

    Args:
        client: The Assets API client to use for fetching components and
            connections.
        config: How to build the graph and generate its formulas. See
            `ComponentGraphConfig`. Defaults to that class's own defaults.
    """
    self._client: AssetsApiClient = client
    self._config: ComponentGraphConfig = (
        config if config is not None else ComponentGraphConfig()
    )
get_component_graph async ¤
get_component_graph(
    microgrid_id: MicrogridId,
) -> MicrogridComponentGraph

Generate a component graph for the given microgrid ID.

PARAMETER DESCRIPTION
microgrid_id

The ID of the microgrid to generate the graph for.

TYPE: MicrogridId

RETURNS DESCRIPTION
MicrogridComponentGraph

The component graph representing the microgrid's electrical components and their connections.
RAISES DESCRIPTION
ValueError

If any component connections could not be loaded.
Source code in src/frequenz/gridpool/_graph_generator.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
async def get_component_graph(
    self, microgrid_id: MicrogridId
) -> MicrogridComponentGraph:
    """Generate a component graph for the given microgrid ID.

    Args:
        microgrid_id: The ID of the microgrid to generate the graph for.

    Returns:
        The component graph representing the microgrid's electrical
            components and their connections.

    Raises:
        ValueError: If any component connections could not be loaded.
    """
    components = await self._client.list_microgrid_electrical_components(
        microgrid_id
    )
    connections = (
        await self._client.list_microgrid_electrical_component_connections(
            microgrid_id
        )
    )

    if any(c is None for c in connections):
        raise ValueError("Failed to load all electrical component connections.")

    breakers = [c for c in components if isinstance(c, Breaker)]
    connected_breakers = [
        b
        for b in breakers
        if any(
            b.id in (c.source, c.destination) for c in connections if c is not None
        )
    ]

    if connected_breakers:
        _logger.warning(
            "The following breakers are connected to other components, "
            + "which is not supported by the component graph generator and may "
            + "lead to graph traversal issues: %s",
            [b.id for b in connected_breakers],
        )
    elif breakers:
        _logger.debug("Dropping unconnected breakers: %s", [b.id for b in breakers])
        components = [c for c in components if not isinstance(c, Breaker)]

    graph = ComponentGraph[
        ElectricalComponent, ComponentConnection, ElectricalComponentId
    ](components, connections, self._config)

    return graph

frequenz.gridpool.FormulaOverrides ¤

Per-formula overrides for the meter/component preference.

Each parameter is None by default, meaning the corresponding formula follows the global prefer_meters_in_component_formulas setting on ComponentGraphConfig. Setting True forces the meter as primary for that formula; False forces the component.

Methods:¤
__init__ ¤
__init__(
    *,
    prefer_meters_in_pv_formula: bool | None = None,
    prefer_meters_in_battery_formula: bool | None = None,
    prefer_meters_in_chp_formula: bool | None = None,
    prefer_meters_in_ev_charger_formula: bool | None = None,
    prefer_meters_in_wind_turbine_formula: (
        bool | None
    ) = None,
    prefer_meters_in_steam_boiler_formula: (
        bool | None
    ) = None
) -> None

Initialize this instance.

PARAMETER DESCRIPTION
prefer_meters_in_pv_formula

Override for pv_formula.

TYPE: bool | None DEFAULT: None

prefer_meters_in_battery_formula

Override for battery_formula.

TYPE: bool | None DEFAULT: None

prefer_meters_in_chp_formula

Override for chp_formula.

TYPE: bool | None DEFAULT: None

prefer_meters_in_ev_charger_formula

Override for ev_charger_formula.

TYPE: bool | None DEFAULT: None

prefer_meters_in_wind_turbine_formula

Override for wind_turbine_formula.

TYPE: bool | None DEFAULT: None

prefer_meters_in_steam_boiler_formula

Override for steam_boiler_formula.

TYPE: bool | None DEFAULT: None