Skip to content

Index

frequenz.sdk.microgrid ¤

Microgrid monitoring and control system.

This package provides a complete suite of data structures and functionality for monitoring and adjusting the state of a microgrid.

Classes¤

frequenz.sdk.microgrid.ComponentGraph ¤

Bases: ABC

Interface for component graph implementations.

Source code in frequenz/sdk/microgrid/_graph.py 
 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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
class ComponentGraph(ABC):
    """Interface for component graph implementations."""

    @abstractmethod
    def components(
        self,
        component_id: Optional[Set[int]] = None,
        component_category: Optional[Set[ComponentCategory]] = None,
    ) -> Set[Component]:
        """Fetch the components of the microgrid.

        Args:
            component_id: filter out any components not matching one of the provided IDs
            component_category: filter out any components not matching one of the
                provided types

        Returns:
            Set of the components currently connected to the microgrid, filtered by
                the provided `component_id` and `component_category` values.
        """

    @abstractmethod
    def connections(
        self,
        start: Optional[Set[int]] = None,
        end: Optional[Set[int]] = None,
    ) -> Set[Connection]:
        """Fetch the connections between microgrid components.

        Args:
            start: filter out any connections whose `start` does not match one of these
                component IDs
            end: filter out any connections whose `end` does not match one of these
                component IDs

        Returns:
            Set of the connections between components in the microgrid, filtered by
                the provided `start`/`end` choices.
        """

    @abstractmethod
    def predecessors(self, component_id: int) -> Set[Component]:
        """Fetch the graph predecessors of the specified component.

        Args:
            component_id: numerical ID of the component whose predecessors should be
                fetched

        Returns:
            Set of IDs of the components that are predecessors of `component_id`,
                i.e. for which there is a connection from each of these components to
                `component_id`.

        Raises:
            KeyError: if the specified `component_id` is not in the graph
        """

    @abstractmethod
    def successors(self, component_id: int) -> Set[Component]:
        """Fetch the graph successors of the specified component.

        Args:
            component_id: numerical ID of the component whose successors should be
                fetched

        Returns:
            Set of IDs of the components that are successors of `component_id`,
                i.e. for which there is a connection from `component_id` to each of
                these components.

        Raises:
            KeyError: if the specified `component_id` is not in the graph
        """

    @abstractmethod
    def is_pv_inverter(self, component: Component) -> bool:
        """Check if the specified component is a PV inverter.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is a PV inverter.
        """

    @abstractmethod
    def is_pv_meter(self, component: Component) -> bool:
        """Check if the specified component is a PV meter.

        This is done by checking if the component has only PV inverters as its
        successors.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is a PV meter.
        """

    @abstractmethod
    def is_pv_chain(self, component: Component) -> bool:
        """Check if the specified component is part of a PV chain.

        A component is part of a PV chain if it is a PV meter or a PV inverter.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is part of a PV chain.
        """

    @abstractmethod
    def is_battery_inverter(self, component: Component) -> bool:
        """Check if the specified component is a battery inverter.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is a battery inverter.
        """

    @abstractmethod
    def is_battery_meter(self, component: Component) -> bool:
        """Check if the specified component is a battery meter.

        This is done by checking if the component has only battery inverters as its
        predecessors.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is a battery meter.
        """

    @abstractmethod
    def is_battery_chain(self, component: Component) -> bool:
        """Check if the specified component is part of a battery chain.

        A component is part of a battery chain if it is a battery meter or a battery
        inverter.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is part of a battery chain.
        """

    @abstractmethod
    def is_ev_charger(self, component: Component) -> bool:
        """Check if the specified component is an EV charger.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is an EV charger.
        """

    @abstractmethod
    def is_ev_charger_meter(self, component: Component) -> bool:
        """Check if the specified component is an EV charger meter.

        This is done by checking if the component has only EV chargers as its
        successors.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is an EV charger meter.
        """

    @abstractmethod
    def is_ev_charger_chain(self, component: Component) -> bool:
        """Check if the specified component is part of an EV charger chain.

        A component is part of an EV charger chain if it is an EV charger meter or an
        EV charger.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is part of an EV charger chain.
        """

    @abstractmethod
    def is_chp(self, component: Component) -> bool:
        """Check if the specified component is a CHP.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is a CHP.
        """

    @abstractmethod
    def is_chp_meter(self, component: Component) -> bool:
        """Check if the specified component is a CHP meter.

        This is done by checking if the component has only CHPs as its successors.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is a CHP meter.
        """

    @abstractmethod
    def is_chp_chain(self, component: Component) -> bool:
        """Check if the specified component is part of a CHP chain.

        A component is part of a CHP chain if it is a CHP meter or a CHP.

        Args:
            component: component to check.

        Returns:
            Whether the specified component is part of a CHP chain.
        """

    @abstractmethod
    def dfs(
        self,
        current_node: Component,
        visited: Set[Component],
        condition: Callable[[Component], bool],
    ) -> Set[Component]:
        """
        Search for components that fulfill the condition in the Graph.

        DFS is used for searching the graph. The graph traversal is stopped
        once a component fulfills the condition.

        Args:
            current_node: The current node to search from.
            visited: The set of visited nodes.
            condition: The condition function to check for.

        Returns:
            A set of component ids where the corresponding components fulfill
            the condition function.
        """
Functions¤
components(component_id=None, component_category=None) abstractmethod ¤

Fetch the components of the microgrid.

PARAMETER DESCRIPTION
component_id

filter out any components not matching one of the provided IDs

TYPE: Optional[Set[int]] DEFAULT: None

component_category

filter out any components not matching one of the provided types

TYPE: Optional[Set[ComponentCategory]] DEFAULT: None

RETURNS DESCRIPTION
Set[Component]

Set of the components currently connected to the microgrid, filtered by the provided component_id and component_category values.
Source code in frequenz/sdk/microgrid/_graph.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@abstractmethod
def components(
    self,
    component_id: Optional[Set[int]] = None,
    component_category: Optional[Set[ComponentCategory]] = None,
) -> Set[Component]:
    """Fetch the components of the microgrid.

    Args:
        component_id: filter out any components not matching one of the provided IDs
        component_category: filter out any components not matching one of the
            provided types

    Returns:
        Set of the components currently connected to the microgrid, filtered by
            the provided `component_id` and `component_category` values.
    """
connections(start=None, end=None) abstractmethod ¤

Fetch the connections between microgrid components.

PARAMETER DESCRIPTION
start

filter out any connections whose start does not match one of these component IDs

TYPE: Optional[Set[int]] DEFAULT: None

end

filter out any connections whose end does not match one of these component IDs

TYPE: Optional[Set[int]] DEFAULT: None

RETURNS DESCRIPTION
Set[Connection]

Set of the connections between components in the microgrid, filtered by the provided start/end choices.
Source code in frequenz/sdk/microgrid/_graph.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@abstractmethod
def connections(
    self,
    start: Optional[Set[int]] = None,
    end: Optional[Set[int]] = None,
) -> Set[Connection]:
    """Fetch the connections between microgrid components.

    Args:
        start: filter out any connections whose `start` does not match one of these
            component IDs
        end: filter out any connections whose `end` does not match one of these
            component IDs

    Returns:
        Set of the connections between components in the microgrid, filtered by
            the provided `start`/`end` choices.
    """
dfs(current_node, visited, condition) abstractmethod ¤

Search for components that fulfill the condition in the Graph.

DFS is used for searching the graph. The graph traversal is stopped once a component fulfills the condition.

PARAMETER DESCRIPTION
current_node

The current node to search from.

TYPE: Component

visited

The set of visited nodes.

TYPE: Set[Component]

condition

The condition function to check for.

TYPE: Callable[[Component], bool]

RETURNS DESCRIPTION
Set[Component]

A set of component ids where the corresponding components fulfill
Set[Component]

the condition function.
Source code in frequenz/sdk/microgrid/_graph.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
@abstractmethod
def dfs(
    self,
    current_node: Component,
    visited: Set[Component],
    condition: Callable[[Component], bool],
) -> Set[Component]:
    """
    Search for components that fulfill the condition in the Graph.

    DFS is used for searching the graph. The graph traversal is stopped
    once a component fulfills the condition.

    Args:
        current_node: The current node to search from.
        visited: The set of visited nodes.
        condition: The condition function to check for.

    Returns:
        A set of component ids where the corresponding components fulfill
        the condition function.
    """
is_battery_chain(component) abstractmethod ¤

Check if the specified component is part of a battery chain.

A component is part of a battery chain if it is a battery meter or a battery inverter.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is part of a battery chain.
Source code in frequenz/sdk/microgrid/_graph.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
@abstractmethod
def is_battery_chain(self, component: Component) -> bool:
    """Check if the specified component is part of a battery chain.

    A component is part of a battery chain if it is a battery meter or a battery
    inverter.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is part of a battery chain.
    """
is_battery_inverter(component) abstractmethod ¤

Check if the specified component is a battery inverter.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is a battery inverter.
Source code in frequenz/sdk/microgrid/_graph.py 
154
155
156
157
158
159
160
161
162
163
@abstractmethod
def is_battery_inverter(self, component: Component) -> bool:
    """Check if the specified component is a battery inverter.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is a battery inverter.
    """
is_battery_meter(component) abstractmethod ¤

Check if the specified component is a battery meter.

This is done by checking if the component has only battery inverters as its predecessors.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is a battery meter.
Source code in frequenz/sdk/microgrid/_graph.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
@abstractmethod
def is_battery_meter(self, component: Component) -> bool:
    """Check if the specified component is a battery meter.

    This is done by checking if the component has only battery inverters as its
    predecessors.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is a battery meter.
    """
is_chp(component) abstractmethod ¤

Check if the specified component is a CHP.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is a CHP.
Source code in frequenz/sdk/microgrid/_graph.py 
232
233
234
235
236
237
238
239
240
241
@abstractmethod
def is_chp(self, component: Component) -> bool:
    """Check if the specified component is a CHP.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is a CHP.
    """
is_chp_chain(component) abstractmethod ¤

Check if the specified component is part of a CHP chain.

A component is part of a CHP chain if it is a CHP meter or a CHP.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is part of a CHP chain.
Source code in frequenz/sdk/microgrid/_graph.py 
256
257
258
259
260
261
262
263
264
265
266
267
@abstractmethod
def is_chp_chain(self, component: Component) -> bool:
    """Check if the specified component is part of a CHP chain.

    A component is part of a CHP chain if it is a CHP meter or a CHP.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is part of a CHP chain.
    """
is_chp_meter(component) abstractmethod ¤

Check if the specified component is a CHP meter.

This is done by checking if the component has only CHPs as its successors.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is a CHP meter.
Source code in frequenz/sdk/microgrid/_graph.py 
243
244
245
246
247
248
249
250
251
252
253
254
@abstractmethod
def is_chp_meter(self, component: Component) -> bool:
    """Check if the specified component is a CHP meter.

    This is done by checking if the component has only CHPs as its successors.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is a CHP meter.
    """
is_ev_charger(component) abstractmethod ¤

Check if the specified component is an EV charger.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is an EV charger.
Source code in frequenz/sdk/microgrid/_graph.py 
193
194
195
196
197
198
199
200
201
202
@abstractmethod
def is_ev_charger(self, component: Component) -> bool:
    """Check if the specified component is an EV charger.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is an EV charger.
    """
is_ev_charger_chain(component) abstractmethod ¤

Check if the specified component is part of an EV charger chain.

A component is part of an EV charger chain if it is an EV charger meter or an EV charger.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is part of an EV charger chain.
Source code in frequenz/sdk/microgrid/_graph.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
@abstractmethod
def is_ev_charger_chain(self, component: Component) -> bool:
    """Check if the specified component is part of an EV charger chain.

    A component is part of an EV charger chain if it is an EV charger meter or an
    EV charger.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is part of an EV charger chain.
    """
is_ev_charger_meter(component) abstractmethod ¤

Check if the specified component is an EV charger meter.

This is done by checking if the component has only EV chargers as its successors.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is an EV charger meter.
Source code in frequenz/sdk/microgrid/_graph.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
@abstractmethod
def is_ev_charger_meter(self, component: Component) -> bool:
    """Check if the specified component is an EV charger meter.

    This is done by checking if the component has only EV chargers as its
    successors.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is an EV charger meter.
    """
is_pv_chain(component) abstractmethod ¤

Check if the specified component is part of a PV chain.

A component is part of a PV chain if it is a PV meter or a PV inverter.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is part of a PV chain.
Source code in frequenz/sdk/microgrid/_graph.py 
141
142
143
144
145
146
147
148
149
150
151
152
@abstractmethod
def is_pv_chain(self, component: Component) -> bool:
    """Check if the specified component is part of a PV chain.

    A component is part of a PV chain if it is a PV meter or a PV inverter.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is part of a PV chain.
    """
is_pv_inverter(component) abstractmethod ¤

Check if the specified component is a PV inverter.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is a PV inverter.
Source code in frequenz/sdk/microgrid/_graph.py 
116
117
118
119
120
121
122
123
124
125
@abstractmethod
def is_pv_inverter(self, component: Component) -> bool:
    """Check if the specified component is a PV inverter.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is a PV inverter.
    """
is_pv_meter(component) abstractmethod ¤

Check if the specified component is a PV meter.

This is done by checking if the component has only PV inverters as its successors.

PARAMETER DESCRIPTION
component

component to check.

TYPE: Component

RETURNS DESCRIPTION
bool

Whether the specified component is a PV meter.
Source code in frequenz/sdk/microgrid/_graph.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
@abstractmethod
def is_pv_meter(self, component: Component) -> bool:
    """Check if the specified component is a PV meter.

    This is done by checking if the component has only PV inverters as its
    successors.

    Args:
        component: component to check.

    Returns:
        Whether the specified component is a PV meter.
    """
predecessors(component_id) abstractmethod ¤

Fetch the graph predecessors of the specified component.

PARAMETER DESCRIPTION
component_id

numerical ID of the component whose predecessors should be fetched

TYPE: int

RETURNS DESCRIPTION
Set[Component]

Set of IDs of the components that are predecessors of component_id, i.e. for which there is a connection from each of these components to component_id.
RAISES DESCRIPTION
KeyError

if the specified component_id is not in the graph
Source code in frequenz/sdk/microgrid/_graph.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
@abstractmethod
def predecessors(self, component_id: int) -> Set[Component]:
    """Fetch the graph predecessors of the specified component.

    Args:
        component_id: numerical ID of the component whose predecessors should be
            fetched

    Returns:
        Set of IDs of the components that are predecessors of `component_id`,
            i.e. for which there is a connection from each of these components to
            `component_id`.

    Raises:
        KeyError: if the specified `component_id` is not in the graph
    """
successors(component_id) abstractmethod ¤

Fetch the graph successors of the specified component.

PARAMETER DESCRIPTION
component_id

numerical ID of the component whose successors should be fetched

TYPE: int

RETURNS DESCRIPTION
Set[Component]

Set of IDs of the components that are successors of component_id, i.e. for which there is a connection from component_id to each of these components.
RAISES DESCRIPTION
KeyError

if the specified component_id is not in the graph
Source code in frequenz/sdk/microgrid/_graph.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@abstractmethod
def successors(self, component_id: int) -> Set[Component]:
    """Fetch the graph successors of the specified component.

    Args:
        component_id: numerical ID of the component whose successors should be
            fetched

    Returns:
        Set of IDs of the components that are successors of `component_id`,
            i.e. for which there is a connection from `component_id` to each of
            these components.

    Raises:
        KeyError: if the specified `component_id` is not in the graph
    """

Functions¤

frequenz.sdk.microgrid.battery_pool(battery_ids=None) ¤

Return the corresponding BatteryPool instance for the given ids.

If a BatteryPool instance for the given ids doesn't exist, a new one is created and returned.

PARAMETER DESCRIPTION
battery_ids

Optional set of IDs of batteries to be managed by the BatteryPool. If not specified, all batteries available in the component graph are used.

TYPE: Set[int] | None DEFAULT: None

RETURNS DESCRIPTION
BatteryPool

A BatteryPool instance.
Source code in frequenz/sdk/microgrid/_data_pipeline.py 
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
def battery_pool(battery_ids: abc.Set[int] | None = None) -> BatteryPool:
    """Return the corresponding BatteryPool instance for the given ids.

    If a BatteryPool instance for the given ids doesn't exist, a new one is
    created and returned.

    Args:
        battery_ids: Optional set of IDs of batteries to be managed by the
            BatteryPool.  If not specified, all batteries available in the
            component graph are used.

    Returns:
        A BatteryPool instance.
    """
    return _get().battery_pool(battery_ids)

frequenz.sdk.microgrid.ev_charger_pool(ev_charger_ids=None) ¤

Return the corresponding EVChargerPool instance for the given ids.

If an EVChargerPool instance for the given ids doesn't exist, a new one is created and returned.

PARAMETER DESCRIPTION
ev_charger_ids

Optional set of IDs of EV Chargers to be managed by the EVChargerPool. If not specified, all EV Chargers available in the component graph are used.

TYPE: set[int] | None DEFAULT: None

RETURNS DESCRIPTION
EVChargerPool

An EVChargerPool instance.
Source code in frequenz/sdk/microgrid/_data_pipeline.py 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
def ev_charger_pool(ev_charger_ids: set[int] | None = None) -> EVChargerPool:
    """Return the corresponding EVChargerPool instance for the given ids.

    If an EVChargerPool instance for the given ids doesn't exist, a new one is
    created and returned.

    Args:
        ev_charger_ids: Optional set of IDs of EV Chargers to be managed by the
            EVChargerPool.  If not specified, all EV Chargers available in the
            component graph are used.

    Returns:
        An EVChargerPool instance.
    """
    return _get().ev_charger_pool(ev_charger_ids)

frequenz.sdk.microgrid.initialize(host, port, resampler_config) async ¤

Initialize the microgrid connection manager and the data pipeline.

PARAMETER DESCRIPTION
host

Host to connect to, to reach the microgrid API.

TYPE: str

port

port to connect to.

TYPE: int

resampler_config

Configuration for the resampling actor.

TYPE: ResamplerConfig

Source code in frequenz/sdk/microgrid/__init__.py 
16
17
18
19
20
21
22
23
24
25
async def initialize(host: str, port: int, resampler_config: ResamplerConfig) -> None:
    """Initialize the microgrid connection manager and the data pipeline.

    Args:
        host: Host to connect to, to reach the microgrid API.
        port: port to connect to.
        resampler_config: Configuration for the resampling actor.
    """
    await connection_manager.initialize(host, port)
    await _data_pipeline.initialize(resampler_config)

frequenz.sdk.microgrid.logical_meter() ¤

Return the logical meter instance.

If a LogicalMeter instance doesn't exist, a new one is created and returned.

RETURNS DESCRIPTION
LogicalMeter

A logical meter instance.
Source code in frequenz/sdk/microgrid/_data_pipeline.py 
299
300
301
302
303
304
305
306
307
def logical_meter() -> LogicalMeter:
    """Return the logical meter instance.

    If a LogicalMeter instance doesn't exist, a new one is created and returned.

    Returns:
        A logical meter instance.
    """
    return _get().logical_meter()