Skip to content

microgrid

frequenz.client.microgrid ¤

Client to connect to the Microgrid API.

This package provides a low-level interface for interacting with the microgrid API.

Classes¤

frequenz.client.microgrid.ApiClient ¤

A microgrid API client.

Source code in frequenz/client/microgrid/_client.py
 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
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
class ApiClient:
    """A microgrid API client."""

    def __init__(
        self,
        server_url: str,
        *,
        channel_options: channel.ChannelOptions = DEFAULT_CHANNEL_OPTIONS,
        retry_strategy: retry.Strategy | None = None,
    ) -> None:
        """Initialize the class instance.

        Args:
            server_url: The location of the microgrid API server in the form of a URL.
                The following format is expected:
                "grpc://hostname{:`port`}{?ssl=`ssl`}",
                where the `port` should be an int between 0 and 65535 (defaulting to
                9090) and `ssl` should be a boolean (defaulting to `false`).
                For example: `grpc://localhost:1090?ssl=true`.
            channel_options: The default options use to create the channel when not
                specified in the URL.
            retry_strategy: The retry strategy to use to reconnect when the connection
                to the streaming method is lost. By default a linear backoff strategy
                is used.
        """
        self._server_url = server_url
        """The location of the microgrid API server as a URL."""

        self.api = microgrid_pb2_grpc.MicrogridStub(
            channel.parse_grpc_uri(server_url, defaults=channel_options)
        )
        """The gRPC stub for the microgrid API."""

        self._broadcasters: dict[int, streaming.GrpcStreamBroadcaster[Any, Any]] = {}
        self._retry_strategy = retry_strategy

    @property
    def server_url(self) -> str:
        """The server location in URL format."""
        return self._server_url

    async def components(self) -> Iterable[Component]:
        """Fetch all the components present in the microgrid.

        Returns:
            Iterator whose elements are all the components in the microgrid.

        Raises:
            ApiClientError: If the are any errors communicating with the Microgrid API,
                most likely a subclass of
                [GrpcError][frequenz.client.microgrid.GrpcError].
        """
        try:
            # grpc.aio is missing types and mypy thinks this is not awaitable,
            # but it is
            component_list = await cast(
                Awaitable[microgrid_pb2.ComponentList],
                self.api.ListComponents(
                    microgrid_pb2.ComponentFilter(),
                    timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
                ),
            )
        except grpc.aio.AioRpcError as grpc_error:
            raise ApiClientError.from_grpc_error(
                server_url=self._server_url,
                operation="ListComponents",
                grpc_error=grpc_error,
            ) from grpc_error

        components_only = filter(
            lambda c: c.category
            is not components_pb2.ComponentCategory.COMPONENT_CATEGORY_SENSOR,
            component_list.components,
        )
        result: Iterable[Component] = map(
            lambda c: Component(
                c.id,
                component_category_from_protobuf(c.category),
                component_type_from_protobuf(c.category, c.inverter),
                component_metadata_from_protobuf(c.category, c.grid),
            ),
            components_only,
        )

        return result

    async def metadata(self) -> Metadata:
        """Fetch the microgrid metadata.

        If there is an error fetching the metadata, the microgrid ID and
        location will be set to None.

        Returns:
            the microgrid metadata.
        """
        microgrid_metadata: microgrid_pb2.MicrogridMetadata | None = None
        try:
            microgrid_metadata = await cast(
                Awaitable[microgrid_pb2.MicrogridMetadata],
                self.api.GetMicrogridMetadata(
                    Empty(),
                    timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
                ),
            )
        except grpc.aio.AioRpcError:
            _logger.exception("The microgrid metadata is not available.")

        if not microgrid_metadata:
            return Metadata()

        location: Location | None = None
        if microgrid_metadata.location:
            location = Location(
                latitude=microgrid_metadata.location.latitude,
                longitude=microgrid_metadata.location.longitude,
            )

        return Metadata(microgrid_id=microgrid_metadata.microgrid_id, location=location)

    async def connections(
        self,
        starts: Set[int] = frozenset(),
        ends: Set[int] = frozenset(),
    ) -> Iterable[Connection]:
        """Fetch the connections between components in the microgrid.

        Args:
            starts: if set and non-empty, only include connections whose start
                value matches one of the provided component IDs
            ends: if set and non-empty, only include connections whose end value
                matches one of the provided component IDs

        Returns:
            Microgrid connections matching the provided start and end filters.

        Raises:
            ApiClientError: If the are any errors communicating with the Microgrid API,
                most likely a subclass of
                [GrpcError][frequenz.client.microgrid.GrpcError].
        """
        connection_filter = microgrid_pb2.ConnectionFilter(starts=starts, ends=ends)
        try:
            valid_components, all_connections = await asyncio.gather(
                self.components(),
                # grpc.aio is missing types and mypy thinks this is not
                # awaitable, but it is
                cast(
                    Awaitable[microgrid_pb2.ConnectionList],
                    self.api.ListConnections(
                        connection_filter,
                        timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
                    ),
                ),
            )
        except grpc.aio.AioRpcError as grpc_error:
            raise ApiClientError.from_grpc_error(
                server_url=self._server_url,
                operation="ListConnections",
                grpc_error=grpc_error,
            ) from grpc_error
        # Filter out the components filtered in `components` method.
        # id=0 is an exception indicating grid component.
        valid_ids = {c.component_id for c in valid_components}
        valid_ids.add(0)

        connections = filter(
            lambda c: (c.start in valid_ids and c.end in valid_ids),
            all_connections.connections,
        )

        result: Iterable[Connection] = map(
            lambda c: Connection(c.start, c.end), connections
        )

        return result

    async def _new_component_data_receiver(
        self,
        *,
        component_id: int,
        expected_category: ComponentCategory,
        transform: Callable[[microgrid_pb2.ComponentData], _ComponentDataT],
        maxsize: int,
    ) -> Receiver[_ComponentDataT]:
        """Return a new broadcaster receiver for a given `component_id`.

        If a broadcaster for the given `component_id` doesn't exist, it creates a new
        one.

        Args:
            component_id: id of the component to get data for.
            expected_category: Category of the component to get data for.
            transform: A method for transforming raw component data into the
                desired output type.
            maxsize: Size of the receiver's buffer.

        Returns:
            The new receiver for the given `component_id`.
        """
        await self._expect_category(
            component_id,
            expected_category,
        )

        broadcaster = self._broadcasters.get(component_id)
        if broadcaster is None:
            broadcaster = streaming.GrpcStreamBroadcaster(
                f"raw-component-data-{component_id}",
                # We need to cast here because grpc says StreamComponentData is
                # a grpc.CallIterator[microgrid_pb2.ComponentData] which is not an
                # AsyncIterator, but it is a grpc.aio.UnaryStreamCall[...,
                # microgrid_pb2.ComponentData], which it is.
                lambda: cast(
                    AsyncIterator[microgrid_pb2.ComponentData],
                    self.api.StreamComponentData(
                        microgrid_pb2.ComponentIdParam(id=component_id)
                    ),
                ),
                transform,
                retry_strategy=self._retry_strategy,
            )
            self._broadcasters[component_id] = broadcaster
        return broadcaster.new_receiver(maxsize=maxsize)

    async def _expect_category(
        self,
        component_id: int,
        expected_category: ComponentCategory,
    ) -> None:
        """Check if the given component_id is of the expected type.

        Raises:
            ValueError: if the given id is unknown or has a different type.

        Args:
            component_id: Component id to check.
            expected_category: Component category that the given id is expected
                to have.
        """
        try:
            comp = next(
                comp
                for comp in await self.components()
                if comp.component_id == component_id
            )
        except StopIteration as exc:
            raise ValueError(
                f"Unable to find component with id {component_id}"
            ) from exc

        if comp.category != expected_category:
            raise ValueError(
                f"Component id {component_id} is a {comp.category.name.lower()}"
                f", not a {expected_category.name.lower()}."
            )

    async def meter_data(  # noqa: DOC502 (ValueError is raised indirectly by _expect_category)
        self,
        component_id: int,
        maxsize: int = RECEIVER_MAX_SIZE,
    ) -> Receiver[MeterData]:
        """Return a channel receiver that provides a `MeterData` stream.

        Raises:
            ValueError: if the given id is unknown or has a different type.

        Args:
            component_id: id of the meter to get data for.
            maxsize: Size of the receiver's buffer.

        Returns:
            A channel receiver that provides realtime meter data.
        """
        return await self._new_component_data_receiver(
            component_id=component_id,
            expected_category=ComponentCategory.METER,
            transform=MeterData.from_proto,
            maxsize=maxsize,
        )

    async def battery_data(  # noqa: DOC502 (ValueError is raised indirectly by _expect_category)
        self,
        component_id: int,
        maxsize: int = RECEIVER_MAX_SIZE,
    ) -> Receiver[BatteryData]:
        """Return a channel receiver that provides a `BatteryData` stream.

        Raises:
            ValueError: if the given id is unknown or has a different type.

        Args:
            component_id: id of the battery to get data for.
            maxsize: Size of the receiver's buffer.

        Returns:
            A channel receiver that provides realtime battery data.
        """
        return await self._new_component_data_receiver(
            component_id=component_id,
            expected_category=ComponentCategory.BATTERY,
            transform=BatteryData.from_proto,
            maxsize=maxsize,
        )

    async def inverter_data(  # noqa: DOC502 (ValueError is raised indirectly by _expect_category)
        self,
        component_id: int,
        maxsize: int = RECEIVER_MAX_SIZE,
    ) -> Receiver[InverterData]:
        """Return a channel receiver that provides an `InverterData` stream.

        Raises:
            ValueError: if the given id is unknown or has a different type.

        Args:
            component_id: id of the inverter to get data for.
            maxsize: Size of the receiver's buffer.

        Returns:
            A channel receiver that provides realtime inverter data.
        """
        return await self._new_component_data_receiver(
            component_id=component_id,
            expected_category=ComponentCategory.INVERTER,
            transform=InverterData.from_proto,
            maxsize=maxsize,
        )

    async def ev_charger_data(  # noqa: DOC502 (ValueError is raised indirectly by _expect_category)
        self,
        component_id: int,
        maxsize: int = RECEIVER_MAX_SIZE,
    ) -> Receiver[EVChargerData]:
        """Return a channel receiver that provides an `EvChargeData` stream.

        Raises:
            ValueError: if the given id is unknown or has a different type.

        Args:
            component_id: id of the ev charger to get data for.
            maxsize: Size of the receiver's buffer.

        Returns:
            A channel receiver that provides realtime ev charger data.
        """
        return await self._new_component_data_receiver(
            component_id=component_id,
            expected_category=ComponentCategory.EV_CHARGER,
            transform=EVChargerData.from_proto,
            maxsize=maxsize,
        )

    async def set_power(self, component_id: int, power_w: float) -> None:
        """Send request to the Microgrid to set power for component.

        If power > 0, then component will be charged with this power.
        If power < 0, then component will be discharged with this power.
        If power == 0, then stop charging or discharging component.


        Args:
            component_id: id of the component to set power.
            power_w: power to set for the component.

        Raises:
            ApiClientError: If the are any errors communicating with the Microgrid API,
                most likely a subclass of
                [GrpcError][frequenz.client.microgrid.GrpcError].
        """
        try:
            await cast(
                Awaitable[Empty],
                self.api.SetPowerActive(
                    microgrid_pb2.SetPowerActiveParam(
                        component_id=component_id, power=power_w
                    ),
                    timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
                ),
            )
        except grpc.aio.AioRpcError as grpc_error:
            raise ApiClientError.from_grpc_error(
                server_url=self._server_url,
                operation="SetPowerActive",
                grpc_error=grpc_error,
            ) from grpc_error

    async def set_reactive_power(  # noqa: DOC502 (raises ApiClientError indirectly)
        self, component_id: int, reactive_power_var: float
    ) -> None:
        """Send request to the Microgrid to set reactive power for component.

        Negative values are for inductive (lagging) power , and positive values are for
        capacitive (leading) power.

        Args:
            component_id: id of the component to set power.
            reactive_power_var: reactive power to set for the component.

        Raises:
            ApiClientError: If the are any errors communicating with the Microgrid API,
                most likely a subclass of
                [GrpcError][frequenz.client.microgrid.GrpcError].
        """
        try:
            await cast(
                Awaitable[Empty],
                self.api.SetPowerReactive(
                    microgrid_pb2.SetPowerReactiveParam(
                        component_id=component_id, power=reactive_power_var
                    ),
                    timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
                ),
            )
        except grpc.aio.AioRpcError as grpc_error:
            raise ApiClientError.from_grpc_error(
                server_url=self._server_url,
                operation="SetPowerReactive",
                grpc_error=grpc_error,
            ) from grpc_error

    async def set_bounds(  # noqa: DOC503 (raises ApiClientError indirectly)
        self,
        component_id: int,
        lower: float,
        upper: float,
    ) -> None:
        """Send `SetBoundsParam`s received from a channel to the Microgrid service.

        Args:
            component_id: ID of the component to set bounds for.
            lower: Lower bound to be set for the component.
            upper: Upper bound to be set for the component.

        Raises:
            ValueError: when upper bound is less than 0, or when lower bound is
                greater than 0.
            ApiClientError: If the are any errors communicating with the Microgrid API,
                most likely a subclass of
                [GrpcError][frequenz.client.microgrid.GrpcError].
        """
        if upper < 0:
            raise ValueError(f"Upper bound {upper} must be greater than or equal to 0.")
        if lower > 0:
            raise ValueError(f"Lower bound {lower} must be less than or equal to 0.")

        target_metric = (
            microgrid_pb2.SetBoundsParam.TargetMetric.TARGET_METRIC_POWER_ACTIVE
        )
        try:
            await cast(
                Awaitable[Timestamp],
                self.api.AddInclusionBounds(
                    microgrid_pb2.SetBoundsParam(
                        component_id=component_id,
                        target_metric=target_metric,
                        bounds=metrics_pb2.Bounds(lower=lower, upper=upper),
                    ),
                    timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
                ),
            )
        except grpc.aio.AioRpcError as grpc_error:
            raise ApiClientError.from_grpc_error(
                server_url=self._server_url,
                operation="AddInclusionBounds",
                grpc_error=grpc_error,
            ) from grpc_error
Attributes¤
api instance-attribute ¤
api = MicrogridStub(
    parse_grpc_uri(server_url, defaults=channel_options)
)

The gRPC stub for the microgrid API.

server_url property ¤
server_url: str

The server location in URL format.

Functions¤
__init__ ¤
__init__(
    server_url: str,
    *,
    channel_options: ChannelOptions = DEFAULT_CHANNEL_OPTIONS,
    retry_strategy: Strategy | None = None
) -> None

Initialize the class instance.

PARAMETER DESCRIPTION
server_url

The location of the microgrid API server in the form of a URL. The following format is expected: "grpc://hostname{:port}{?ssl=ssl}", where the port should be an int between 0 and 65535 (defaulting to 9090) and ssl should be a boolean (defaulting to false). For example: grpc://localhost:1090?ssl=true.

TYPE: str

channel_options

The default options use to create the channel when not specified in the URL.

TYPE: ChannelOptions DEFAULT: DEFAULT_CHANNEL_OPTIONS

retry_strategy

The retry strategy to use to reconnect when the connection to the streaming method is lost. By default a linear backoff strategy is used.

TYPE: Strategy | None DEFAULT: None

Source code in frequenz/client/microgrid/_client.py
def __init__(
    self,
    server_url: str,
    *,
    channel_options: channel.ChannelOptions = DEFAULT_CHANNEL_OPTIONS,
    retry_strategy: retry.Strategy | None = None,
) -> None:
    """Initialize the class instance.

    Args:
        server_url: The location of the microgrid API server in the form of a URL.
            The following format is expected:
            "grpc://hostname{:`port`}{?ssl=`ssl`}",
            where the `port` should be an int between 0 and 65535 (defaulting to
            9090) and `ssl` should be a boolean (defaulting to `false`).
            For example: `grpc://localhost:1090?ssl=true`.
        channel_options: The default options use to create the channel when not
            specified in the URL.
        retry_strategy: The retry strategy to use to reconnect when the connection
            to the streaming method is lost. By default a linear backoff strategy
            is used.
    """
    self._server_url = server_url
    """The location of the microgrid API server as a URL."""

    self.api = microgrid_pb2_grpc.MicrogridStub(
        channel.parse_grpc_uri(server_url, defaults=channel_options)
    )
    """The gRPC stub for the microgrid API."""

    self._broadcasters: dict[int, streaming.GrpcStreamBroadcaster[Any, Any]] = {}
    self._retry_strategy = retry_strategy
battery_data async ¤
battery_data(
    component_id: int, maxsize: int = RECEIVER_MAX_SIZE
) -> Receiver[BatteryData]

Return a channel receiver that provides a BatteryData stream.

RAISES DESCRIPTION
ValueError

if the given id is unknown or has a different type.

PARAMETER DESCRIPTION
component_id

id of the battery to get data for.

TYPE: int

maxsize

Size of the receiver's buffer.

TYPE: int DEFAULT: RECEIVER_MAX_SIZE

RETURNS DESCRIPTION
Receiver[BatteryData]

A channel receiver that provides realtime battery data.

Source code in frequenz/client/microgrid/_client.py
async def battery_data(  # noqa: DOC502 (ValueError is raised indirectly by _expect_category)
    self,
    component_id: int,
    maxsize: int = RECEIVER_MAX_SIZE,
) -> Receiver[BatteryData]:
    """Return a channel receiver that provides a `BatteryData` stream.

    Raises:
        ValueError: if the given id is unknown or has a different type.

    Args:
        component_id: id of the battery to get data for.
        maxsize: Size of the receiver's buffer.

    Returns:
        A channel receiver that provides realtime battery data.
    """
    return await self._new_component_data_receiver(
        component_id=component_id,
        expected_category=ComponentCategory.BATTERY,
        transform=BatteryData.from_proto,
        maxsize=maxsize,
    )
components async ¤
components() -> Iterable[Component]

Fetch all the components present in the microgrid.

RETURNS DESCRIPTION
Iterable[Component]

Iterator whose elements are all the components in the microgrid.

RAISES DESCRIPTION
ApiClientError

If the are any errors communicating with the Microgrid API, most likely a subclass of GrpcError.

Source code in frequenz/client/microgrid/_client.py
async def components(self) -> Iterable[Component]:
    """Fetch all the components present in the microgrid.

    Returns:
        Iterator whose elements are all the components in the microgrid.

    Raises:
        ApiClientError: If the are any errors communicating with the Microgrid API,
            most likely a subclass of
            [GrpcError][frequenz.client.microgrid.GrpcError].
    """
    try:
        # grpc.aio is missing types and mypy thinks this is not awaitable,
        # but it is
        component_list = await cast(
            Awaitable[microgrid_pb2.ComponentList],
            self.api.ListComponents(
                microgrid_pb2.ComponentFilter(),
                timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
            ),
        )
    except grpc.aio.AioRpcError as grpc_error:
        raise ApiClientError.from_grpc_error(
            server_url=self._server_url,
            operation="ListComponents",
            grpc_error=grpc_error,
        ) from grpc_error

    components_only = filter(
        lambda c: c.category
        is not components_pb2.ComponentCategory.COMPONENT_CATEGORY_SENSOR,
        component_list.components,
    )
    result: Iterable[Component] = map(
        lambda c: Component(
            c.id,
            component_category_from_protobuf(c.category),
            component_type_from_protobuf(c.category, c.inverter),
            component_metadata_from_protobuf(c.category, c.grid),
        ),
        components_only,
    )

    return result
connections async ¤
connections(
    starts: Set[int] = frozenset(),
    ends: Set[int] = frozenset(),
) -> Iterable[Connection]

Fetch the connections between components in the microgrid.

PARAMETER DESCRIPTION
starts

if set and non-empty, only include connections whose start value matches one of the provided component IDs

TYPE: Set[int] DEFAULT: frozenset()

ends

if set and non-empty, only include connections whose end value matches one of the provided component IDs

TYPE: Set[int] DEFAULT: frozenset()

RETURNS DESCRIPTION
Iterable[Connection]

Microgrid connections matching the provided start and end filters.

RAISES DESCRIPTION
ApiClientError

If the are any errors communicating with the Microgrid API, most likely a subclass of GrpcError.

Source code in frequenz/client/microgrid/_client.py
async def connections(
    self,
    starts: Set[int] = frozenset(),
    ends: Set[int] = frozenset(),
) -> Iterable[Connection]:
    """Fetch the connections between components in the microgrid.

    Args:
        starts: if set and non-empty, only include connections whose start
            value matches one of the provided component IDs
        ends: if set and non-empty, only include connections whose end value
            matches one of the provided component IDs

    Returns:
        Microgrid connections matching the provided start and end filters.

    Raises:
        ApiClientError: If the are any errors communicating with the Microgrid API,
            most likely a subclass of
            [GrpcError][frequenz.client.microgrid.GrpcError].
    """
    connection_filter = microgrid_pb2.ConnectionFilter(starts=starts, ends=ends)
    try:
        valid_components, all_connections = await asyncio.gather(
            self.components(),
            # grpc.aio is missing types and mypy thinks this is not
            # awaitable, but it is
            cast(
                Awaitable[microgrid_pb2.ConnectionList],
                self.api.ListConnections(
                    connection_filter,
                    timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
                ),
            ),
        )
    except grpc.aio.AioRpcError as grpc_error:
        raise ApiClientError.from_grpc_error(
            server_url=self._server_url,
            operation="ListConnections",
            grpc_error=grpc_error,
        ) from grpc_error
    # Filter out the components filtered in `components` method.
    # id=0 is an exception indicating grid component.
    valid_ids = {c.component_id for c in valid_components}
    valid_ids.add(0)

    connections = filter(
        lambda c: (c.start in valid_ids and c.end in valid_ids),
        all_connections.connections,
    )

    result: Iterable[Connection] = map(
        lambda c: Connection(c.start, c.end), connections
    )

    return result
ev_charger_data async ¤
ev_charger_data(
    component_id: int, maxsize: int = RECEIVER_MAX_SIZE
) -> Receiver[EVChargerData]

Return a channel receiver that provides an EvChargeData stream.

RAISES DESCRIPTION
ValueError

if the given id is unknown or has a different type.

PARAMETER DESCRIPTION
component_id

id of the ev charger to get data for.

TYPE: int

maxsize

Size of the receiver's buffer.

TYPE: int DEFAULT: RECEIVER_MAX_SIZE

RETURNS DESCRIPTION
Receiver[EVChargerData]

A channel receiver that provides realtime ev charger data.

Source code in frequenz/client/microgrid/_client.py
async def ev_charger_data(  # noqa: DOC502 (ValueError is raised indirectly by _expect_category)
    self,
    component_id: int,
    maxsize: int = RECEIVER_MAX_SIZE,
) -> Receiver[EVChargerData]:
    """Return a channel receiver that provides an `EvChargeData` stream.

    Raises:
        ValueError: if the given id is unknown or has a different type.

    Args:
        component_id: id of the ev charger to get data for.
        maxsize: Size of the receiver's buffer.

    Returns:
        A channel receiver that provides realtime ev charger data.
    """
    return await self._new_component_data_receiver(
        component_id=component_id,
        expected_category=ComponentCategory.EV_CHARGER,
        transform=EVChargerData.from_proto,
        maxsize=maxsize,
    )
inverter_data async ¤
inverter_data(
    component_id: int, maxsize: int = RECEIVER_MAX_SIZE
) -> Receiver[InverterData]

Return a channel receiver that provides an InverterData stream.

RAISES DESCRIPTION
ValueError

if the given id is unknown or has a different type.

PARAMETER DESCRIPTION
component_id

id of the inverter to get data for.

TYPE: int

maxsize

Size of the receiver's buffer.

TYPE: int DEFAULT: RECEIVER_MAX_SIZE

RETURNS DESCRIPTION
Receiver[InverterData]

A channel receiver that provides realtime inverter data.

Source code in frequenz/client/microgrid/_client.py
async def inverter_data(  # noqa: DOC502 (ValueError is raised indirectly by _expect_category)
    self,
    component_id: int,
    maxsize: int = RECEIVER_MAX_SIZE,
) -> Receiver[InverterData]:
    """Return a channel receiver that provides an `InverterData` stream.

    Raises:
        ValueError: if the given id is unknown or has a different type.

    Args:
        component_id: id of the inverter to get data for.
        maxsize: Size of the receiver's buffer.

    Returns:
        A channel receiver that provides realtime inverter data.
    """
    return await self._new_component_data_receiver(
        component_id=component_id,
        expected_category=ComponentCategory.INVERTER,
        transform=InverterData.from_proto,
        maxsize=maxsize,
    )
metadata async ¤
metadata() -> Metadata

Fetch the microgrid metadata.

If there is an error fetching the metadata, the microgrid ID and location will be set to None.

RETURNS DESCRIPTION
Metadata

the microgrid metadata.

Source code in frequenz/client/microgrid/_client.py
async def metadata(self) -> Metadata:
    """Fetch the microgrid metadata.

    If there is an error fetching the metadata, the microgrid ID and
    location will be set to None.

    Returns:
        the microgrid metadata.
    """
    microgrid_metadata: microgrid_pb2.MicrogridMetadata | None = None
    try:
        microgrid_metadata = await cast(
            Awaitable[microgrid_pb2.MicrogridMetadata],
            self.api.GetMicrogridMetadata(
                Empty(),
                timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
            ),
        )
    except grpc.aio.AioRpcError:
        _logger.exception("The microgrid metadata is not available.")

    if not microgrid_metadata:
        return Metadata()

    location: Location | None = None
    if microgrid_metadata.location:
        location = Location(
            latitude=microgrid_metadata.location.latitude,
            longitude=microgrid_metadata.location.longitude,
        )

    return Metadata(microgrid_id=microgrid_metadata.microgrid_id, location=location)
meter_data async ¤
meter_data(
    component_id: int, maxsize: int = RECEIVER_MAX_SIZE
) -> Receiver[MeterData]

Return a channel receiver that provides a MeterData stream.

RAISES DESCRIPTION
ValueError

if the given id is unknown or has a different type.

PARAMETER DESCRIPTION
component_id

id of the meter to get data for.

TYPE: int

maxsize

Size of the receiver's buffer.

TYPE: int DEFAULT: RECEIVER_MAX_SIZE

RETURNS DESCRIPTION
Receiver[MeterData]

A channel receiver that provides realtime meter data.

Source code in frequenz/client/microgrid/_client.py
async def meter_data(  # noqa: DOC502 (ValueError is raised indirectly by _expect_category)
    self,
    component_id: int,
    maxsize: int = RECEIVER_MAX_SIZE,
) -> Receiver[MeterData]:
    """Return a channel receiver that provides a `MeterData` stream.

    Raises:
        ValueError: if the given id is unknown or has a different type.

    Args:
        component_id: id of the meter to get data for.
        maxsize: Size of the receiver's buffer.

    Returns:
        A channel receiver that provides realtime meter data.
    """
    return await self._new_component_data_receiver(
        component_id=component_id,
        expected_category=ComponentCategory.METER,
        transform=MeterData.from_proto,
        maxsize=maxsize,
    )
set_bounds async ¤
set_bounds(
    component_id: int, lower: float, upper: float
) -> None

Send SetBoundsParams received from a channel to the Microgrid service.

PARAMETER DESCRIPTION
component_id

ID of the component to set bounds for.

TYPE: int

lower

Lower bound to be set for the component.

TYPE: float

upper

Upper bound to be set for the component.

TYPE: float

RAISES DESCRIPTION
ValueError

when upper bound is less than 0, or when lower bound is greater than 0.

ApiClientError

If the are any errors communicating with the Microgrid API, most likely a subclass of GrpcError.

Source code in frequenz/client/microgrid/_client.py
async def set_bounds(  # noqa: DOC503 (raises ApiClientError indirectly)
    self,
    component_id: int,
    lower: float,
    upper: float,
) -> None:
    """Send `SetBoundsParam`s received from a channel to the Microgrid service.

    Args:
        component_id: ID of the component to set bounds for.
        lower: Lower bound to be set for the component.
        upper: Upper bound to be set for the component.

    Raises:
        ValueError: when upper bound is less than 0, or when lower bound is
            greater than 0.
        ApiClientError: If the are any errors communicating with the Microgrid API,
            most likely a subclass of
            [GrpcError][frequenz.client.microgrid.GrpcError].
    """
    if upper < 0:
        raise ValueError(f"Upper bound {upper} must be greater than or equal to 0.")
    if lower > 0:
        raise ValueError(f"Lower bound {lower} must be less than or equal to 0.")

    target_metric = (
        microgrid_pb2.SetBoundsParam.TargetMetric.TARGET_METRIC_POWER_ACTIVE
    )
    try:
        await cast(
            Awaitable[Timestamp],
            self.api.AddInclusionBounds(
                microgrid_pb2.SetBoundsParam(
                    component_id=component_id,
                    target_metric=target_metric,
                    bounds=metrics_pb2.Bounds(lower=lower, upper=upper),
                ),
                timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
            ),
        )
    except grpc.aio.AioRpcError as grpc_error:
        raise ApiClientError.from_grpc_error(
            server_url=self._server_url,
            operation="AddInclusionBounds",
            grpc_error=grpc_error,
        ) from grpc_error
set_power async ¤
set_power(component_id: int, power_w: float) -> None

Send request to the Microgrid to set power for component.

If power > 0, then component will be charged with this power. If power < 0, then component will be discharged with this power. If power == 0, then stop charging or discharging component.

PARAMETER DESCRIPTION
component_id

id of the component to set power.

TYPE: int

power_w

power to set for the component.

TYPE: float

RAISES DESCRIPTION
ApiClientError

If the are any errors communicating with the Microgrid API, most likely a subclass of GrpcError.

Source code in frequenz/client/microgrid/_client.py
async def set_power(self, component_id: int, power_w: float) -> None:
    """Send request to the Microgrid to set power for component.

    If power > 0, then component will be charged with this power.
    If power < 0, then component will be discharged with this power.
    If power == 0, then stop charging or discharging component.


    Args:
        component_id: id of the component to set power.
        power_w: power to set for the component.

    Raises:
        ApiClientError: If the are any errors communicating with the Microgrid API,
            most likely a subclass of
            [GrpcError][frequenz.client.microgrid.GrpcError].
    """
    try:
        await cast(
            Awaitable[Empty],
            self.api.SetPowerActive(
                microgrid_pb2.SetPowerActiveParam(
                    component_id=component_id, power=power_w
                ),
                timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
            ),
        )
    except grpc.aio.AioRpcError as grpc_error:
        raise ApiClientError.from_grpc_error(
            server_url=self._server_url,
            operation="SetPowerActive",
            grpc_error=grpc_error,
        ) from grpc_error
set_reactive_power async ¤
set_reactive_power(
    component_id: int, reactive_power_var: float
) -> None

Send request to the Microgrid to set reactive power for component.

Negative values are for inductive (lagging) power , and positive values are for capacitive (leading) power.

PARAMETER DESCRIPTION
component_id

id of the component to set power.

TYPE: int

reactive_power_var

reactive power to set for the component.

TYPE: float

RAISES DESCRIPTION
ApiClientError

If the are any errors communicating with the Microgrid API, most likely a subclass of GrpcError.

Source code in frequenz/client/microgrid/_client.py
async def set_reactive_power(  # noqa: DOC502 (raises ApiClientError indirectly)
    self, component_id: int, reactive_power_var: float
) -> None:
    """Send request to the Microgrid to set reactive power for component.

    Negative values are for inductive (lagging) power , and positive values are for
    capacitive (leading) power.

    Args:
        component_id: id of the component to set power.
        reactive_power_var: reactive power to set for the component.

    Raises:
        ApiClientError: If the are any errors communicating with the Microgrid API,
            most likely a subclass of
            [GrpcError][frequenz.client.microgrid.GrpcError].
    """
    try:
        await cast(
            Awaitable[Empty],
            self.api.SetPowerReactive(
                microgrid_pb2.SetPowerReactiveParam(
                    component_id=component_id, power=reactive_power_var
                ),
                timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
            ),
        )
    except grpc.aio.AioRpcError as grpc_error:
        raise ApiClientError.from_grpc_error(
            server_url=self._server_url,
            operation="SetPowerReactive",
            grpc_error=grpc_error,
        ) from grpc_error

frequenz.client.microgrid.ApiClientError ¤

Bases: Exception

There was an error in an API client.

To simplify retrying, errors are classified as retryable, or not. Retryable errors might succeed if retried, while permanent errors won't. When uncertain, errors are assumed to be retryable.

The following sub-classes are available:

Source code in frequenz/client/base/exception.py
class ApiClientError(Exception):
    """There was an error in an API client.

    To simplify retrying, errors are classified as
    [retryable][frequenz.client.base.exception.ApiClientError.is_retryable], or not.
    Retryable errors might succeed if retried, while permanent errors won't. When
    uncertain, errors are assumed to be retryable.

    The following sub-classes are available:

    - [GrpcError][frequenz.client.base.exception.GrpcError]: A gRPC operation failed.
    """

    def __init__(
        self,
        *,
        server_url: str,
        operation: str,
        description: str,
        retryable: bool,
    ) -> None:
        """Create a new instance.

        Args:
            server_url: The URL of the server that returned the error.
            operation: The operation that caused the error.
            description: A human-readable description of the error.
            retryable: Whether retrying the operation might succeed.
        """
        super().__init__(
            f"Failed calling {operation!r} on {server_url!r}: {description}"
        )

        self.server_url = server_url
        """The URL of the server that returned the error."""

        self.operation = operation
        """The operation that caused the error."""

        self.description = description
        """The human-readable description of the error."""

        self.is_retryable = retryable
        """Whether retrying the operation might succeed."""

    @classmethod
    def from_grpc_error(
        cls,
        *,
        server_url: str,
        operation: str,
        grpc_error: AioRpcError,
    ) -> GrpcError:
        """Create an instance of the appropriate subclass from a gRPC error.

        Args:
            server_url: The URL of the server that returned the error.
            operation: The operation that caused the error.
            grpc_error: The gRPC error to convert.

        Returns:
            An instance of
                [GrpcError][frequenz.client.base.exception.GrpcError] if the gRPC status
                is not recognized, or an appropriate subclass if it is.
        """

        class Ctor(Protocol):
            """A protocol for the constructor of a subclass of `GrpcError`."""

            def __call__(
                self, *, server_url: str, operation: str, grpc_error: AioRpcError
            ) -> GrpcError: ...

        grpc_status_map: dict[grpc.StatusCode, Ctor] = {
            grpc.StatusCode.CANCELLED: OperationCancelled,
            grpc.StatusCode.UNKNOWN: UnknownError,
            grpc.StatusCode.INVALID_ARGUMENT: InvalidArgument,
            grpc.StatusCode.DEADLINE_EXCEEDED: OperationTimedOut,
            grpc.StatusCode.NOT_FOUND: EntityNotFound,
            grpc.StatusCode.ALREADY_EXISTS: EntityAlreadyExists,
            grpc.StatusCode.PERMISSION_DENIED: PermissionDenied,
            grpc.StatusCode.RESOURCE_EXHAUSTED: ResourceExhausted,
            grpc.StatusCode.FAILED_PRECONDITION: OperationPreconditionFailed,
            grpc.StatusCode.ABORTED: OperationAborted,
            grpc.StatusCode.OUT_OF_RANGE: OperationOutOfRange,
            grpc.StatusCode.UNIMPLEMENTED: OperationNotImplemented,
            grpc.StatusCode.INTERNAL: InternalError,
            grpc.StatusCode.UNAVAILABLE: ServiceUnavailable,
            grpc.StatusCode.DATA_LOSS: DataLoss,
            grpc.StatusCode.UNAUTHENTICATED: OperationUnauthenticated,
        }

        if ctor := grpc_status_map.get(grpc_error.code()):
            return ctor(
                server_url=server_url, operation=operation, grpc_error=grpc_error
            )
        return UnrecognizedGrpcStatus(
            server_url=server_url,
            operation=operation,
            grpc_error=grpc_error,
        )
Attributes¤
description instance-attribute ¤
description = description

The human-readable description of the error.

is_retryable instance-attribute ¤
is_retryable = retryable

Whether retrying the operation might succeed.

operation instance-attribute ¤
operation = operation

The operation that caused the error.

server_url instance-attribute ¤
server_url = server_url

The URL of the server that returned the error.

Functions¤
__init__ ¤
__init__(
    *,
    server_url: str,
    operation: str,
    description: str,
    retryable: bool
) -> None

Create a new instance.

PARAMETER DESCRIPTION
server_url

The URL of the server that returned the error.

TYPE: str

operation

The operation that caused the error.

TYPE: str

description

A human-readable description of the error.

TYPE: str

retryable

Whether retrying the operation might succeed.

TYPE: bool

Source code in frequenz/client/base/exception.py
def __init__(
    self,
    *,
    server_url: str,
    operation: str,
    description: str,
    retryable: bool,
) -> None:
    """Create a new instance.

    Args:
        server_url: The URL of the server that returned the error.
        operation: The operation that caused the error.
        description: A human-readable description of the error.
        retryable: Whether retrying the operation might succeed.
    """
    super().__init__(
        f"Failed calling {operation!r} on {server_url!r}: {description}"
    )

    self.server_url = server_url
    """The URL of the server that returned the error."""

    self.operation = operation
    """The operation that caused the error."""

    self.description = description
    """The human-readable description of the error."""

    self.is_retryable = retryable
    """Whether retrying the operation might succeed."""
from_grpc_error classmethod ¤
from_grpc_error(
    *,
    server_url: str,
    operation: str,
    grpc_error: AioRpcError
) -> GrpcError

Create an instance of the appropriate subclass from a gRPC error.

PARAMETER DESCRIPTION
server_url

The URL of the server that returned the error.

TYPE: str

operation

The operation that caused the error.

TYPE: str

grpc_error

The gRPC error to convert.

TYPE: AioRpcError

RETURNS DESCRIPTION
GrpcError

An instance of GrpcError if the gRPC status is not recognized, or an appropriate subclass if it is.

Source code in frequenz/client/base/exception.py
@classmethod
def from_grpc_error(
    cls,
    *,
    server_url: str,
    operation: str,
    grpc_error: AioRpcError,
) -> GrpcError:
    """Create an instance of the appropriate subclass from a gRPC error.

    Args:
        server_url: The URL of the server that returned the error.
        operation: The operation that caused the error.
        grpc_error: The gRPC error to convert.

    Returns:
        An instance of
            [GrpcError][frequenz.client.base.exception.GrpcError] if the gRPC status
            is not recognized, or an appropriate subclass if it is.
    """

    class Ctor(Protocol):
        """A protocol for the constructor of a subclass of `GrpcError`."""

        def __call__(
            self, *, server_url: str, operation: str, grpc_error: AioRpcError
        ) -> GrpcError: ...

    grpc_status_map: dict[grpc.StatusCode, Ctor] = {
        grpc.StatusCode.CANCELLED: OperationCancelled,
        grpc.StatusCode.UNKNOWN: UnknownError,
        grpc.StatusCode.INVALID_ARGUMENT: InvalidArgument,
        grpc.StatusCode.DEADLINE_EXCEEDED: OperationTimedOut,
        grpc.StatusCode.NOT_FOUND: EntityNotFound,
        grpc.StatusCode.ALREADY_EXISTS: EntityAlreadyExists,
        grpc.StatusCode.PERMISSION_DENIED: PermissionDenied,
        grpc.StatusCode.RESOURCE_EXHAUSTED: ResourceExhausted,
        grpc.StatusCode.FAILED_PRECONDITION: OperationPreconditionFailed,
        grpc.StatusCode.ABORTED: OperationAborted,
        grpc.StatusCode.OUT_OF_RANGE: OperationOutOfRange,
        grpc.StatusCode.UNIMPLEMENTED: OperationNotImplemented,
        grpc.StatusCode.INTERNAL: InternalError,
        grpc.StatusCode.UNAVAILABLE: ServiceUnavailable,
        grpc.StatusCode.DATA_LOSS: DataLoss,
        grpc.StatusCode.UNAUTHENTICATED: OperationUnauthenticated,
    }

    if ctor := grpc_status_map.get(grpc_error.code()):
        return ctor(
            server_url=server_url, operation=operation, grpc_error=grpc_error
        )
    return UnrecognizedGrpcStatus(
        server_url=server_url,
        operation=operation,
        grpc_error=grpc_error,
    )

frequenz.client.microgrid.BatteryComponentState ¤

Bases: Enum

Component states of a battery.

Source code in frequenz/client/microgrid/_component_states.py
class BatteryComponentState(Enum):
    """Component states of a battery."""

    UNSPECIFIED = battery_pb2.ComponentState.COMPONENT_STATE_UNSPECIFIED
    """Unspecified component state."""

    OFF = battery_pb2.ComponentState.COMPONENT_STATE_OFF
    """The battery is switched off."""

    IDLE = battery_pb2.ComponentState.COMPONENT_STATE_IDLE
    """The battery is idle."""

    CHARGING = battery_pb2.ComponentState.COMPONENT_STATE_CHARGING
    """The battery is consuming electrical energy."""

    DISCHARGING = battery_pb2.ComponentState.COMPONENT_STATE_DISCHARGING
    """The battery is generating electrical energy."""

    ERROR = battery_pb2.ComponentState.COMPONENT_STATE_ERROR
    """The battery is in a faulty state."""

    LOCKED = battery_pb2.ComponentState.COMPONENT_STATE_LOCKED
    """The battery is online, but currently unavailable.

    Possibly due to a pre-scheduled maintenance, or waiting for a resource to be loaded.
    """

    SWITCHING_ON = battery_pb2.ComponentState.COMPONENT_STATE_SWITCHING_ON
    """
    The battery is starting up and needs some time to become fully operational.
    """

    SWITCHING_OFF = battery_pb2.ComponentState.COMPONENT_STATE_SWITCHING_OFF
    """The battery is switching off and needs some time to fully shut down."""

    UNKNOWN = battery_pb2.ComponentState.COMPONENT_STATE_UNKNOWN
    """A state is provided by the component, but it is not one of the above states."""

    @classmethod
    def from_pb(cls, state: battery_pb2.ComponentState.ValueType) -> Self:
        """Convert a protobuf state value to this enum.

        Args:
            state: The protobuf component state to convert.

        Returns:
            The enum value corresponding to the protobuf message.
        """
        try:
            return cls(state)
        except ValueError:
            return cls(cls.UNKNOWN)
Attributes¤
CHARGING class-attribute instance-attribute ¤
CHARGING = COMPONENT_STATE_CHARGING

The battery is consuming electrical energy.

DISCHARGING class-attribute instance-attribute ¤
DISCHARGING = COMPONENT_STATE_DISCHARGING

The battery is generating electrical energy.

ERROR class-attribute instance-attribute ¤
ERROR = COMPONENT_STATE_ERROR

The battery is in a faulty state.

IDLE class-attribute instance-attribute ¤
IDLE = COMPONENT_STATE_IDLE

The battery is idle.

LOCKED class-attribute instance-attribute ¤
LOCKED = COMPONENT_STATE_LOCKED

The battery is online, but currently unavailable.

Possibly due to a pre-scheduled maintenance, or waiting for a resource to be loaded.

OFF class-attribute instance-attribute ¤
OFF = COMPONENT_STATE_OFF

The battery is switched off.

SWITCHING_OFF class-attribute instance-attribute ¤
SWITCHING_OFF = COMPONENT_STATE_SWITCHING_OFF

The battery is switching off and needs some time to fully shut down.

SWITCHING_ON class-attribute instance-attribute ¤
SWITCHING_ON = COMPONENT_STATE_SWITCHING_ON

The battery is starting up and needs some time to become fully operational.

UNKNOWN class-attribute instance-attribute ¤
UNKNOWN = COMPONENT_STATE_UNKNOWN

A state is provided by the component, but it is not one of the above states.

UNSPECIFIED class-attribute instance-attribute ¤
UNSPECIFIED = COMPONENT_STATE_UNSPECIFIED

Unspecified component state.

Functions¤
from_pb classmethod ¤
from_pb(state: ValueType) -> Self

Convert a protobuf state value to this enum.

PARAMETER DESCRIPTION
state

The protobuf component state to convert.

TYPE: ValueType

RETURNS DESCRIPTION
Self

The enum value corresponding to the protobuf message.

Source code in frequenz/client/microgrid/_component_states.py
@classmethod
def from_pb(cls, state: battery_pb2.ComponentState.ValueType) -> Self:
    """Convert a protobuf state value to this enum.

    Args:
        state: The protobuf component state to convert.

    Returns:
        The enum value corresponding to the protobuf message.
    """
    try:
        return cls(state)
    except ValueError:
        return cls(cls.UNKNOWN)

frequenz.client.microgrid.BatteryData dataclass ¤

Bases: ComponentData

A wrapper class for holding battery data.

Source code in frequenz/client/microgrid/_component_data.py
@dataclass(frozen=True)
class BatteryData(ComponentData):  # pylint: disable=too-many-instance-attributes
    """A wrapper class for holding battery data."""

    soc: float
    """Battery's overall SoC in percent (%)."""

    soc_lower_bound: float
    """The SoC below which discharge commands will be blocked by the system,
        in percent (%).
    """

    soc_upper_bound: float
    """The SoC above which charge commands will be blocked by the system,
        in percent (%).
    """

    capacity: float
    """The capacity of the battery in Wh (Watt-hour)."""

    power_inclusion_lower_bound: float
    """Lower inclusion bound for battery power in watts.

    This is the lower limit of the range within which power requests are allowed for the
    battery.

    See [`frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds`][] and
    [`frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds`][] for more
    details.
    """

    power_exclusion_lower_bound: float
    """Lower exclusion bound for battery power in watts.

    This is the lower limit of the range within which power requests are not allowed for
    the battery.

    See [`frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds`][] and
    [`frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds`][] for more
    details.
    """

    power_inclusion_upper_bound: float
    """Upper inclusion bound for battery power in watts.

    This is the upper limit of the range within which power requests are allowed for the
    battery.

    See [`frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds`][] and
    [`frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds`][] for more
    details.
    """

    power_exclusion_upper_bound: float
    """Upper exclusion bound for battery power in watts.

    This is the upper limit of the range within which power requests are not allowed for
    the battery.

    See [`frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds`][] and
    [`frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds`][] for more
    details.
    """

    temperature: float
    """The (average) temperature reported by the battery, in Celsius (°C)."""

    relay_state: BatteryRelayState
    """State of the battery relay."""

    component_state: BatteryComponentState
    """State of the battery."""

    errors: list[BatteryError]
    """List of errors in protobuf struct."""

    @classmethod
    def from_proto(cls, raw: microgrid_pb2.ComponentData) -> Self:
        """Create BatteryData from a protobuf message.

        Args:
            raw: raw component data as decoded from the wire.

        Returns:
            Instance of BatteryData created from the protobuf message.
        """
        raw_power = raw.battery.data.dc.power
        battery_data = cls(
            component_id=raw.id,
            timestamp=raw.ts.ToDatetime(tzinfo=timezone.utc),
            soc=raw.battery.data.soc.avg,
            soc_lower_bound=raw.battery.data.soc.system_inclusion_bounds.lower,
            soc_upper_bound=raw.battery.data.soc.system_inclusion_bounds.upper,
            capacity=raw.battery.properties.capacity,
            power_inclusion_lower_bound=raw_power.system_inclusion_bounds.lower,
            power_exclusion_lower_bound=raw_power.system_exclusion_bounds.lower,
            power_inclusion_upper_bound=raw_power.system_inclusion_bounds.upper,
            power_exclusion_upper_bound=raw_power.system_exclusion_bounds.upper,
            temperature=raw.battery.data.temperature.avg,
            relay_state=BatteryRelayState.from_pb(raw.battery.state.relay_state),
            component_state=BatteryComponentState.from_pb(
                raw.battery.state.component_state
            ),
            errors=[BatteryError.from_pb(e) for e in raw.battery.errors],
        )
        battery_data._set_raw(raw=raw)
        return battery_data
Attributes¤
capacity instance-attribute ¤
capacity: float

The capacity of the battery in Wh (Watt-hour).

component_id instance-attribute ¤
component_id: int

The ID identifying this component in the microgrid.

component_state instance-attribute ¤
component_state: BatteryComponentState

State of the battery.

errors instance-attribute ¤
errors: list[BatteryError]

List of errors in protobuf struct.

power_exclusion_lower_bound instance-attribute ¤
power_exclusion_lower_bound: float

Lower exclusion bound for battery power in watts.

This is the lower limit of the range within which power requests are not allowed for the battery.

See frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds and frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds for more details.

power_exclusion_upper_bound instance-attribute ¤
power_exclusion_upper_bound: float

Upper exclusion bound for battery power in watts.

This is the upper limit of the range within which power requests are not allowed for the battery.

See frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds and frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds for more details.

power_inclusion_lower_bound instance-attribute ¤
power_inclusion_lower_bound: float

Lower inclusion bound for battery power in watts.

This is the lower limit of the range within which power requests are allowed for the battery.

See frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds and frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds for more details.

power_inclusion_upper_bound instance-attribute ¤
power_inclusion_upper_bound: float

Upper inclusion bound for battery power in watts.

This is the upper limit of the range within which power requests are allowed for the battery.

See frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds and frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds for more details.

raw class-attribute instance-attribute ¤
raw: ComponentData | None = field(default=None, init=False)

Raw component data as decoded from the wire.

relay_state instance-attribute ¤
relay_state: BatteryRelayState

State of the battery relay.

soc instance-attribute ¤
soc: float

Battery's overall SoC in percent (%).

soc_lower_bound instance-attribute ¤
soc_lower_bound: float

The SoC below which discharge commands will be blocked by the system, in percent (%).

soc_upper_bound instance-attribute ¤
soc_upper_bound: float

The SoC above which charge commands will be blocked by the system, in percent (%).

temperature instance-attribute ¤
temperature: float

The (average) temperature reported by the battery, in Celsius (°C).

timestamp instance-attribute ¤
timestamp: datetime

The timestamp of when the data was measured.

Functions¤
from_proto classmethod ¤
from_proto(raw: ComponentData) -> Self

Create BatteryData from a protobuf message.

PARAMETER DESCRIPTION
raw

raw component data as decoded from the wire.

TYPE: ComponentData

RETURNS DESCRIPTION
Self

Instance of BatteryData created from the protobuf message.

Source code in frequenz/client/microgrid/_component_data.py
@classmethod
def from_proto(cls, raw: microgrid_pb2.ComponentData) -> Self:
    """Create BatteryData from a protobuf message.

    Args:
        raw: raw component data as decoded from the wire.

    Returns:
        Instance of BatteryData created from the protobuf message.
    """
    raw_power = raw.battery.data.dc.power
    battery_data = cls(
        component_id=raw.id,
        timestamp=raw.ts.ToDatetime(tzinfo=timezone.utc),
        soc=raw.battery.data.soc.avg,
        soc_lower_bound=raw.battery.data.soc.system_inclusion_bounds.lower,
        soc_upper_bound=raw.battery.data.soc.system_inclusion_bounds.upper,
        capacity=raw.battery.properties.capacity,
        power_inclusion_lower_bound=raw_power.system_inclusion_bounds.lower,
        power_exclusion_lower_bound=raw_power.system_exclusion_bounds.lower,
        power_inclusion_upper_bound=raw_power.system_inclusion_bounds.upper,
        power_exclusion_upper_bound=raw_power.system_exclusion_bounds.upper,
        temperature=raw.battery.data.temperature.avg,
        relay_state=BatteryRelayState.from_pb(raw.battery.state.relay_state),
        component_state=BatteryComponentState.from_pb(
            raw.battery.state.component_state
        ),
        errors=[BatteryError.from_pb(e) for e in raw.battery.errors],
    )
    battery_data._set_raw(raw=raw)
    return battery_data

frequenz.client.microgrid.BatteryError dataclass ¤

A battery error.

Source code in frequenz/client/microgrid/_component_error.py
@dataclass(frozen=True, kw_only=True)
class BatteryError:
    """A battery error."""

    code: BatteryErrorCode = BatteryErrorCode.UNSPECIFIED
    """The error code."""

    level: ErrorLevel = ErrorLevel.UNSPECIFIED
    """The error level."""

    message: str = ""
    """The error message."""

    @classmethod
    def from_pb(cls, raw: battery_pb2.Error) -> Self:
        """Create a new instance using a protobuf message to get the values.

        Args:
            raw: The protobuf message to get the values from.

        Returns:
            The new instance with the values from the protobuf message.
        """
        return cls(
            code=BatteryErrorCode.from_pb(raw.code),
            level=ErrorLevel.from_pb(raw.level),
            message=raw.msg,
        )
Attributes¤
code class-attribute instance-attribute ¤

The error code.

level class-attribute instance-attribute ¤

The error level.

message class-attribute instance-attribute ¤
message: str = ''

The error message.

Functions¤
from_pb classmethod ¤
from_pb(raw: Error) -> Self

Create a new instance using a protobuf message to get the values.

PARAMETER DESCRIPTION
raw

The protobuf message to get the values from.

TYPE: Error

RETURNS DESCRIPTION
Self

The new instance with the values from the protobuf message.

Source code in frequenz/client/microgrid/_component_error.py
@classmethod
def from_pb(cls, raw: battery_pb2.Error) -> Self:
    """Create a new instance using a protobuf message to get the values.

    Args:
        raw: The protobuf message to get the values from.

    Returns:
        The new instance with the values from the protobuf message.
    """
    return cls(
        code=BatteryErrorCode.from_pb(raw.code),
        level=ErrorLevel.from_pb(raw.level),
        message=raw.msg,
    )

frequenz.client.microgrid.BatteryErrorCode ¤

Bases: Enum

Battery error code.

Source code in frequenz/client/microgrid/_component_error.py
class BatteryErrorCode(Enum):
    """Battery error code."""

    UNSPECIFIED = battery_pb2.ErrorCode.ERROR_CODE_UNSPECIFIED
    """Unspecified battery error code."""

    HIGH_CURRENT_CHARGE = battery_pb2.ErrorCode.ERROR_CODE_HIGH_CURRENT_CHARGE
    """Charge current is too high."""

    HIGH_CURRENT_DISCHARGE = battery_pb2.ErrorCode.ERROR_CODE_HIGH_CURRENT_DISCHARGE
    """Discharge current is too high."""

    HIGH_VOLTAGE = battery_pb2.ErrorCode.ERROR_CODE_HIGH_VOLTAGE
    """Voltage is too high."""

    LOW_VOLTAGE = battery_pb2.ErrorCode.ERROR_CODE_LOW_VOLTAGE
    """Voltage is too low."""

    HIGH_TEMPERATURE = battery_pb2.ErrorCode.ERROR_CODE_HIGH_TEMPERATURE
    """Temperature is too high."""

    LOW_TEMPERATURE = battery_pb2.ErrorCode.ERROR_CODE_LOW_TEMPERATURE
    """Temperature is too low."""

    HIGH_HUMIDITY = battery_pb2.ErrorCode.ERROR_CODE_HIGH_HUMIDITY
    """Humidity is too high."""

    EXCEEDED_SOP_CHARGE = battery_pb2.ErrorCode.ERROR_CODE_EXCEEDED_SOP_CHARGE
    """Charge current has exceeded component bounds."""

    EXCEEDED_SOP_DISCHARGE = battery_pb2.ErrorCode.ERROR_CODE_EXCEEDED_SOP_DISCHARGE
    """Discharge current has exceeded component bounds."""

    SYSTEM_IMBALANCE = battery_pb2.ErrorCode.ERROR_CODE_SYSTEM_IMBALANCE
    """The battery blocks are not balanced with respect to each other."""

    LOW_SOH = battery_pb2.ErrorCode.ERROR_CODE_LOW_SOH
    """The State of health is low."""

    BLOCK_ERROR = battery_pb2.ErrorCode.ERROR_CODE_BLOCK_ERROR
    """One or more battery blocks have failed."""

    CONTROLLER_ERROR = battery_pb2.ErrorCode.ERROR_CODE_CONTROLLER_ERROR
    """The battery controller has failed."""

    RELAY_ERROR = battery_pb2.ErrorCode.ERROR_CODE_RELAY_ERROR
    """The battery's DC relays have failed."""

    RELAY_CYCLE_LIMIT_REACHED = (
        battery_pb2.ErrorCode.ERROR_CODE_RELAY_CYCLE_LIMIT_REACHED
    )
    """The battery's DC relays have reached the cycles limit in its lifetime specifications."""

    FUSE_ERROR = battery_pb2.ErrorCode.ERROR_CODE_FUSE_ERROR
    """The battery's fuse has failed."""

    EXTERNAL_POWER_SWITCH_ERROR = (
        battery_pb2.ErrorCode.ERROR_CODE_EXTERNAL_POWER_SWITCH_ERROR
    )
    """The eternal power switch has failed."""

    PRECHARGE_ERROR = battery_pb2.ErrorCode.ERROR_CODE_PRECHARGE_ERROR
    """The precharge operation has failed."""

    SYSTEM_PLAUSIBILITY_ERROR = (
        battery_pb2.ErrorCode.ERROR_CODE_SYSTEM_PLAUSIBILITY_ERROR
    )
    """System plausibility checks have failed."""

    SYSTEM_UNDERVOLTAGE_SHUTDOWN = (
        battery_pb2.ErrorCode.ERROR_CODE_SYSTEM_UNDERVOLTAGE_SHUTDOWN
    )
    """System shut down due to extremely low voltage."""

    CALIBRATION_NEEDED = battery_pb2.ErrorCode.ERROR_CODE_CALIBRATION_NEEDED
    """The battery requires a calibration to reset its measurements."""

    @classmethod
    def from_pb(cls, code: battery_pb2.ErrorCode.ValueType) -> Self:
        """Convert a protobuf error code value to this enum.

        Args:
            code: The protobuf error code to convert.

        Returns:
            The enum value corresponding to the protobuf message.
        """
        try:
            return cls(code)
        except ValueError:
            return cls(cls.UNSPECIFIED)
Attributes¤
BLOCK_ERROR class-attribute instance-attribute ¤
BLOCK_ERROR = ERROR_CODE_BLOCK_ERROR

One or more battery blocks have failed.

CALIBRATION_NEEDED class-attribute instance-attribute ¤
CALIBRATION_NEEDED = ERROR_CODE_CALIBRATION_NEEDED

The battery requires a calibration to reset its measurements.

CONTROLLER_ERROR class-attribute instance-attribute ¤
CONTROLLER_ERROR = ERROR_CODE_CONTROLLER_ERROR

The battery controller has failed.

EXCEEDED_SOP_CHARGE class-attribute instance-attribute ¤
EXCEEDED_SOP_CHARGE = ERROR_CODE_EXCEEDED_SOP_CHARGE

Charge current has exceeded component bounds.

EXCEEDED_SOP_DISCHARGE class-attribute instance-attribute ¤
EXCEEDED_SOP_DISCHARGE = ERROR_CODE_EXCEEDED_SOP_DISCHARGE

Discharge current has exceeded component bounds.

EXTERNAL_POWER_SWITCH_ERROR class-attribute instance-attribute ¤
EXTERNAL_POWER_SWITCH_ERROR = (
    ERROR_CODE_EXTERNAL_POWER_SWITCH_ERROR
)

The eternal power switch has failed.

FUSE_ERROR class-attribute instance-attribute ¤
FUSE_ERROR = ERROR_CODE_FUSE_ERROR

The battery's fuse has failed.

HIGH_CURRENT_CHARGE class-attribute instance-attribute ¤
HIGH_CURRENT_CHARGE = ERROR_CODE_HIGH_CURRENT_CHARGE

Charge current is too high.

HIGH_CURRENT_DISCHARGE class-attribute instance-attribute ¤
HIGH_CURRENT_DISCHARGE = ERROR_CODE_HIGH_CURRENT_DISCHARGE

Discharge current is too high.

HIGH_HUMIDITY class-attribute instance-attribute ¤
HIGH_HUMIDITY = ERROR_CODE_HIGH_HUMIDITY

Humidity is too high.

HIGH_TEMPERATURE class-attribute instance-attribute ¤
HIGH_TEMPERATURE = ERROR_CODE_HIGH_TEMPERATURE

Temperature is too high.

HIGH_VOLTAGE class-attribute instance-attribute ¤
HIGH_VOLTAGE = ERROR_CODE_HIGH_VOLTAGE

Voltage is too high.

LOW_SOH class-attribute instance-attribute ¤
LOW_SOH = ERROR_CODE_LOW_SOH

The State of health is low.

LOW_TEMPERATURE class-attribute instance-attribute ¤
LOW_TEMPERATURE = ERROR_CODE_LOW_TEMPERATURE

Temperature is too low.

LOW_VOLTAGE class-attribute instance-attribute ¤
LOW_VOLTAGE = ERROR_CODE_LOW_VOLTAGE

Voltage is too low.

PRECHARGE_ERROR class-attribute instance-attribute ¤
PRECHARGE_ERROR = ERROR_CODE_PRECHARGE_ERROR

The precharge operation has failed.

RELAY_CYCLE_LIMIT_REACHED class-attribute instance-attribute ¤
RELAY_CYCLE_LIMIT_REACHED = (
    ERROR_CODE_RELAY_CYCLE_LIMIT_REACHED
)

The battery's DC relays have reached the cycles limit in its lifetime specifications.

RELAY_ERROR class-attribute instance-attribute ¤
RELAY_ERROR = ERROR_CODE_RELAY_ERROR

The battery's DC relays have failed.

SYSTEM_IMBALANCE class-attribute instance-attribute ¤
SYSTEM_IMBALANCE = ERROR_CODE_SYSTEM_IMBALANCE

The battery blocks are not balanced with respect to each other.

SYSTEM_PLAUSIBILITY_ERROR class-attribute instance-attribute ¤
SYSTEM_PLAUSIBILITY_ERROR = (
    ERROR_CODE_SYSTEM_PLAUSIBILITY_ERROR
)

System plausibility checks have failed.

SYSTEM_UNDERVOLTAGE_SHUTDOWN class-attribute instance-attribute ¤
SYSTEM_UNDERVOLTAGE_SHUTDOWN = (
    ERROR_CODE_SYSTEM_UNDERVOLTAGE_SHUTDOWN
)

System shut down due to extremely low voltage.

UNSPECIFIED class-attribute instance-attribute ¤
UNSPECIFIED = ERROR_CODE_UNSPECIFIED

Unspecified battery error code.

Functions¤
from_pb classmethod ¤
from_pb(code: ValueType) -> Self

Convert a protobuf error code value to this enum.

PARAMETER DESCRIPTION
code

The protobuf error code to convert.

TYPE: ValueType

RETURNS DESCRIPTION
Self

The enum value corresponding to the protobuf message.

Source code in frequenz/client/microgrid/_component_error.py
@classmethod
def from_pb(cls, code: battery_pb2.ErrorCode.ValueType) -> Self:
    """Convert a protobuf error code value to this enum.

    Args:
        code: The protobuf error code to convert.

    Returns:
        The enum value corresponding to the protobuf message.
    """
    try:
        return cls(code)
    except ValueError:
        return cls(cls.UNSPECIFIED)

frequenz.client.microgrid.BatteryRelayState ¤

Bases: Enum

Relay states of a battery.

Source code in frequenz/client/microgrid/_component_states.py
class BatteryRelayState(Enum):
    """Relay states of a battery."""

    UNSPECIFIED = battery_pb2.RelayState.RELAY_STATE_UNSPECIFIED
    """Unspecified relay state."""

    OPENED = battery_pb2.RelayState.RELAY_STATE_OPENED
    """The relays are open, and the DC power line to the inverter is disconnected."""

    PRECHARGING = battery_pb2.RelayState.RELAY_STATE_PRECHARGING
    """The relays are closing, and the DC power line to the inverter is being connected."""

    CLOSED = battery_pb2.RelayState.RELAY_STATE_CLOSED
    """The relays are closed, and the DC power line to the inverter is connected."""

    ERROR = battery_pb2.RelayState.RELAY_STATE_ERROR
    """The relays are in an error state."""

    LOCKED = battery_pb2.RelayState.RELAY_STATE_LOCKED
    """The relays are locked, and should be available to accept commands shortly."""

    @classmethod
    def from_pb(cls, state: battery_pb2.RelayState.ValueType) -> Self:
        """Convert a protobuf state value to this enum.

        Args:
            state: The protobuf component state to convert.

        Returns:
            The enum value corresponding to the protobuf message.
        """
        try:
            return cls(state)
        except ValueError:
            return cls(cls.UNSPECIFIED)
Attributes¤
CLOSED class-attribute instance-attribute ¤
CLOSED = RELAY_STATE_CLOSED

The relays are closed, and the DC power line to the inverter is connected.

ERROR class-attribute instance-attribute ¤
ERROR = RELAY_STATE_ERROR

The relays are in an error state.

LOCKED class-attribute instance-attribute ¤
LOCKED = RELAY_STATE_LOCKED

The relays are locked, and should be available to accept commands shortly.

OPENED class-attribute instance-attribute ¤
OPENED = RELAY_STATE_OPENED

The relays are open, and the DC power line to the inverter is disconnected.

PRECHARGING class-attribute instance-attribute ¤
PRECHARGING = RELAY_STATE_PRECHARGING

The relays are closing, and the DC power line to the inverter is being connected.

UNSPECIFIED class-attribute instance-attribute ¤
UNSPECIFIED = RELAY_STATE_UNSPECIFIED

Unspecified relay state.

Functions¤
from_pb classmethod ¤
from_pb(state: ValueType) -> Self

Convert a protobuf state value to this enum.

PARAMETER DESCRIPTION
state

The protobuf component state to convert.

TYPE: ValueType

RETURNS DESCRIPTION
Self

The enum value corresponding to the protobuf message.

Source code in frequenz/client/microgrid/_component_states.py
@classmethod
def from_pb(cls, state: battery_pb2.RelayState.ValueType) -> Self:
    """Convert a protobuf state value to this enum.

    Args:
        state: The protobuf component state to convert.

    Returns:
        The enum value corresponding to the protobuf message.
    """
    try:
        return cls(state)
    except ValueError:
        return cls(cls.UNSPECIFIED)

frequenz.client.microgrid.ClientNotConnected ¤

Bases: ApiClientError

The client is not connected to the server.

Source code in frequenz/client/base/exception.py
class ClientNotConnected(ApiClientError):
    """The client is not connected to the server."""

    def __init__(self, *, server_url: str, operation: str) -> None:
        """Create a new instance.

        Args:
            server_url: The URL of the server that returned the error.
            operation: The operation that caused the error.
        """
        super().__init__(
            server_url=server_url,
            operation=operation,
            description="The client is not connected to the server",
            retryable=True,
        )
Attributes¤
description instance-attribute ¤
description = description

The human-readable description of the error.

is_retryable instance-attribute ¤
is_retryable = retryable

Whether retrying the operation might succeed.

operation instance-attribute ¤
operation = operation

The operation that caused the error.

server_url instance-attribute ¤
server_url = server_url

The URL of the server that returned the error.

Functions¤
__init__ ¤
__init__(*, server_url: str, operation: str) -> None

Create a new instance.

PARAMETER DESCRIPTION
server_url

The URL of the server that returned the error.

TYPE: str

operation

The operation that caused the error.

TYPE: str

Source code in frequenz/client/base/exception.py
def __init__(self, *, server_url: str, operation: str) -> None:
    """Create a new instance.

    Args:
        server_url: The URL of the server that returned the error.
        operation: The operation that caused the error.
    """
    super().__init__(
        server_url=server_url,
        operation=operation,
        description="The client is not connected to the server",
        retryable=True,
    )
from_grpc_error classmethod ¤
from_grpc_error(
    *,
    server_url: str,
    operation: str,
    grpc_error: AioRpcError
) -> GrpcError

Create an instance of the appropriate subclass from a gRPC error.

PARAMETER DESCRIPTION
server_url

The URL of the server that returned the error.

TYPE: str

operation

The operation that caused the error.

TYPE: str

grpc_error

The gRPC error to convert.

TYPE: AioRpcError

RETURNS DESCRIPTION
GrpcError

An instance of GrpcError if the gRPC status is not recognized, or an appropriate subclass if it is.

Source code in frequenz/client/base/exception.py
@classmethod
def from_grpc_error(
    cls,
    *,
    server_url: str,
    operation: str,
    grpc_error: AioRpcError,
) -> GrpcError:
    """Create an instance of the appropriate subclass from a gRPC error.

    Args:
        server_url: The URL of the server that returned the error.
        operation: The operation that caused the error.
        grpc_error: The gRPC error to convert.

    Returns:
        An instance of
            [GrpcError][frequenz.client.base.exception.GrpcError] if the gRPC status
            is not recognized, or an appropriate subclass if it is.
    """

    class Ctor(Protocol):
        """A protocol for the constructor of a subclass of `GrpcError`."""

        def __call__(
            self, *, server_url: str, operation: str, grpc_error: AioRpcError
        ) -> GrpcError: ...

    grpc_status_map: dict[grpc.StatusCode, Ctor] = {
        grpc.StatusCode.CANCELLED: OperationCancelled,
        grpc.StatusCode.UNKNOWN: UnknownError,
        grpc.StatusCode.INVALID_ARGUMENT: InvalidArgument,
        grpc.StatusCode.DEADLINE_EXCEEDED: OperationTimedOut,
        grpc.StatusCode.NOT_FOUND: EntityNotFound,
        grpc.StatusCode.ALREADY_EXISTS: EntityAlreadyExists,
        grpc.StatusCode.PERMISSION_DENIED: PermissionDenied,
        grpc.StatusCode.RESOURCE_EXHAUSTED: ResourceExhausted,
        grpc.StatusCode.FAILED_PRECONDITION: OperationPreconditionFailed,
        grpc.StatusCode.ABORTED: OperationAborted,
        grpc.StatusCode.OUT_OF_RANGE: OperationOutOfRange,
        grpc.StatusCode.UNIMPLEMENTED: OperationNotImplemented,
        grpc.StatusCode.INTERNAL: InternalError,
        grpc.StatusCode.UNAVAILABLE: ServiceUnavailable,
        grpc.StatusCode.DATA_LOSS: DataLoss,
        grpc.StatusCode.UNAUTHENTICATED: OperationUnauthenticated,
    }

    if ctor := grpc_status_map.get(grpc_error.code()):
        return ctor(
            server_url=server_url, operation=operation, grpc_error=grpc_error
        )
    return UnrecognizedGrpcStatus(
        server_url=server_url,
        operation=operation,
        grpc_error=grpc_error,
    )

frequenz.client.microgrid.Component dataclass ¤

Metadata for a single microgrid component.

Source code in frequenz/client/microgrid/_component.py
@dataclass(frozen=True)
class Component:
    """Metadata for a single microgrid component."""

    component_id: int
    """The ID of this component."""

    category: ComponentCategory
    """The category of this component."""

    type: ComponentType | None = None
    """The type of this component."""

    metadata: ComponentMetadata | None = None
    """The metadata of this component."""

    def is_valid(self) -> bool:
        """Check if this instance contains valid data.

        Returns:
            `True` if `id > 0` and `type` is a valid `ComponentCategory`, or if `id
                == 0` and `type` is `GRID`, `False` otherwise
        """
        return (
            self.component_id > 0 and any(t == self.category for t in ComponentCategory)
        ) or (self.component_id == 0 and self.category == ComponentCategory.GRID)

    def __hash__(self) -> int:
        """Compute a hash of this instance, obtained by hashing the `component_id` field.

        Returns:
            Hash of this instance.
        """
        return hash(self.component_id)
Attributes¤
category instance-attribute ¤

The category of this component.

component_id instance-attribute ¤
component_id: int

The ID of this component.

metadata class-attribute instance-attribute ¤
metadata: ComponentMetadata | None = None

The metadata of this component.

type class-attribute instance-attribute ¤
type: ComponentType | None = None

The type of this component.

Functions¤
__hash__ ¤
__hash__() -> int

Compute a hash of this instance, obtained by hashing the component_id field.

RETURNS DESCRIPTION
int

Hash of this instance.

Source code in frequenz/client/microgrid/_component.py
def __hash__(self) -> int:
    """Compute a hash of this instance, obtained by hashing the `component_id` field.

    Returns:
        Hash of this instance.
    """
    return hash(self.component_id)
is_valid ¤
is_valid() -> bool

Check if this instance contains valid data.

RETURNS DESCRIPTION
bool

True if id > 0 and type is a valid ComponentCategory, or if id == 0 and type is GRID, False otherwise

Source code in frequenz/client/microgrid/_component.py
def is_valid(self) -> bool:
    """Check if this instance contains valid data.

    Returns:
        `True` if `id > 0` and `type` is a valid `ComponentCategory`, or if `id
            == 0` and `type` is `GRID`, `False` otherwise
    """
    return (
        self.component_id > 0 and any(t == self.category for t in ComponentCategory)
    ) or (self.component_id == 0 and self.category == ComponentCategory.GRID)

frequenz.client.microgrid.ComponentCategory ¤

Bases: Enum

Possible types of microgrid component.

Source code in frequenz/client/microgrid/_component.py
class ComponentCategory(Enum):
    """Possible types of microgrid component."""

    NONE = components_pb2.ComponentCategory.COMPONENT_CATEGORY_UNSPECIFIED
    """Unspecified component category."""

    GRID = components_pb2.ComponentCategory.COMPONENT_CATEGORY_GRID
    """Grid component."""

    METER = components_pb2.ComponentCategory.COMPONENT_CATEGORY_METER
    """Meter component."""

    INVERTER = components_pb2.ComponentCategory.COMPONENT_CATEGORY_INVERTER
    """Inverter component."""

    BATTERY = components_pb2.ComponentCategory.COMPONENT_CATEGORY_BATTERY
    """Battery component."""

    EV_CHARGER = components_pb2.ComponentCategory.COMPONENT_CATEGORY_EV_CHARGER
    """EV charger component."""

    CHP = components_pb2.ComponentCategory.COMPONENT_CATEGORY_CHP
    """CHP component."""
Attributes¤
BATTERY class-attribute instance-attribute ¤
BATTERY = COMPONENT_CATEGORY_BATTERY

Battery component.

CHP class-attribute instance-attribute ¤
CHP = COMPONENT_CATEGORY_CHP

CHP component.

EV_CHARGER class-attribute instance-attribute ¤
EV_CHARGER = COMPONENT_CATEGORY_EV_CHARGER

EV charger component.

GRID class-attribute instance-attribute ¤
GRID = COMPONENT_CATEGORY_GRID

Grid component.

INVERTER class-attribute instance-attribute ¤
INVERTER = COMPONENT_CATEGORY_INVERTER

Inverter component.

METER class-attribute instance-attribute ¤
METER = COMPONENT_CATEGORY_METER

Meter component.

NONE class-attribute instance-attribute ¤
NONE = COMPONENT_CATEGORY_UNSPECIFIED

Unspecified component category.

frequenz.client.microgrid.ComponentData dataclass ¤

Bases: ABC

A private base class for strongly typed component data classes.

Source code in frequenz/client/microgrid/_component_data.py
@dataclass(frozen=True)
class ComponentData(ABC):
    """A private base class for strongly typed component data classes."""

    component_id: int
    """The ID identifying this component in the microgrid."""

    timestamp: datetime
    """The timestamp of when the data was measured."""

    # The `raw` attribute is excluded from the constructor as it can only be provided
    # when instantiating `ComponentData` using the `from_proto` method, which reads
    # data from a protobuf message. The whole protobuf message is stored as the `raw`
    # attribute. When `ComponentData` is not instantiated from a protobuf message,
    # i.e. using the constructor, `raw` will be set to `None`.
    raw: microgrid_pb2.ComponentData | None = field(default=None, init=False)
    """Raw component data as decoded from the wire."""

    def _set_raw(self, raw: microgrid_pb2.ComponentData) -> None:
        """Store raw protobuf message.

        It is preferred to keep the dataclasses immutable (frozen) and make the `raw`
            attribute read-only, which is why the approach of writing to `__dict__`
            was used, instead of mutating the `self.raw = raw` attribute directly.

        Args:
            raw: raw component data as decoded from the wire.
        """
        self.__dict__["raw"] = raw

    @classmethod
    @abstractmethod
    def from_proto(cls, raw: microgrid_pb2.ComponentData) -> Self:
        """Create ComponentData from a protobuf message.

        Args:
            raw: raw component data as decoded from the wire.

        Returns:
            The instance created from the protobuf message.
        """
Attributes¤
component_id instance-attribute ¤
component_id: int

The ID identifying this component in the microgrid.

raw class-attribute instance-attribute ¤
raw: ComponentData | None = field(default=None, init=False)

Raw component data as decoded from the wire.

timestamp instance-attribute ¤
timestamp: datetime

The timestamp of when the data was measured.

Functions¤
from_proto abstractmethod classmethod ¤
from_proto(raw: ComponentData) -> Self

Create ComponentData from a protobuf message.

PARAMETER DESCRIPTION
raw

raw component data as decoded from the wire.

TYPE: ComponentData

RETURNS DESCRIPTION
Self

The instance created from the protobuf message.

Source code in frequenz/client/microgrid/_component_data.py
@classmethod
@abstractmethod
def from_proto(cls, raw: microgrid_pb2.ComponentData) -> Self:
    """Create ComponentData from a protobuf message.

    Args:
        raw: raw component data as decoded from the wire.

    Returns:
        The instance created from the protobuf message.
    """

frequenz.client.microgrid.ComponentMetadata dataclass ¤

Base class for component metadata classes.

Source code in frequenz/client/microgrid/_component.py
@dataclass(frozen=True)
class ComponentMetadata:
    """Base class for component metadata classes."""

    fuse: Fuse | None = None
    """The fuse at the grid connection point."""
Attributes¤
fuse class-attribute instance-attribute ¤
fuse: Fuse | None = None

The fuse at the grid connection point.

frequenz.client.microgrid.ComponentMetricId ¤

Bases: Enum

An enum representing the various metrics available in the microgrid.

Source code in frequenz/client/microgrid/_component.py
class ComponentMetricId(Enum):
    """An enum representing the various metrics available in the microgrid."""

    ACTIVE_POWER = "active_power"
    """Active power."""

    ACTIVE_POWER_PHASE_1 = "active_power_phase_1"
    """Active power in phase 1."""
    ACTIVE_POWER_PHASE_2 = "active_power_phase_2"
    """Active power in phase 2."""
    ACTIVE_POWER_PHASE_3 = "active_power_phase_3"
    """Active power in phase 3."""

    REACTIVE_POWER = "reactive_power"
    """Reactive power."""

    REACTIVE_POWER_PHASE_1 = "reactive_power_phase_1"
    """Reactive power in phase 1."""
    REACTIVE_POWER_PHASE_2 = "reactive_power_phase_2"
    """Reactive power in phase 2."""
    REACTIVE_POWER_PHASE_3 = "reactive_power_phase_3"
    """Reactive power in phase 3."""

    CURRENT_PHASE_1 = "current_phase_1"
    """Current in phase 1."""
    CURRENT_PHASE_2 = "current_phase_2"
    """Current in phase 2."""
    CURRENT_PHASE_3 = "current_phase_3"
    """Current in phase 3."""

    VOLTAGE_PHASE_1 = "voltage_phase_1"
    """Voltage in phase 1."""
    VOLTAGE_PHASE_2 = "voltage_phase_2"
    """Voltage in phase 2."""
    VOLTAGE_PHASE_3 = "voltage_phase_3"
    """Voltage in phase 3."""

    FREQUENCY = "frequency"

    SOC = "soc"
    """State of charge."""
    SOC_LOWER_BOUND = "soc_lower_bound"
    """Lower bound of state of charge."""
    SOC_UPPER_BOUND = "soc_upper_bound"
    """Upper bound of state of charge."""
    CAPACITY = "capacity"
    """Capacity."""

    POWER_INCLUSION_LOWER_BOUND = "power_inclusion_lower_bound"
    """Power inclusion lower bound."""
    POWER_EXCLUSION_LOWER_BOUND = "power_exclusion_lower_bound"
    """Power exclusion lower bound."""
    POWER_EXCLUSION_UPPER_BOUND = "power_exclusion_upper_bound"
    """Power exclusion upper bound."""
    POWER_INCLUSION_UPPER_BOUND = "power_inclusion_upper_bound"
    """Power inclusion upper bound."""

    ACTIVE_POWER_INCLUSION_LOWER_BOUND = "active_power_inclusion_lower_bound"
    """Active power inclusion lower bound."""
    ACTIVE_POWER_EXCLUSION_LOWER_BOUND = "active_power_exclusion_lower_bound"
    """Active power exclusion lower bound."""
    ACTIVE_POWER_EXCLUSION_UPPER_BOUND = "active_power_exclusion_upper_bound"
    """Active power exclusion upper bound."""
    ACTIVE_POWER_INCLUSION_UPPER_BOUND = "active_power_inclusion_upper_bound"
    """Active power inclusion upper bound."""

    TEMPERATURE = "temperature"
    """Temperature."""
Attributes¤
ACTIVE_POWER class-attribute instance-attribute ¤
ACTIVE_POWER = 'active_power'

Active power.

ACTIVE_POWER_EXCLUSION_LOWER_BOUND class-attribute instance-attribute ¤
ACTIVE_POWER_EXCLUSION_LOWER_BOUND = (
    "active_power_exclusion_lower_bound"
)

Active power exclusion lower bound.

ACTIVE_POWER_EXCLUSION_UPPER_BOUND class-attribute instance-attribute ¤
ACTIVE_POWER_EXCLUSION_UPPER_BOUND = (
    "active_power_exclusion_upper_bound"
)

Active power exclusion upper bound.

ACTIVE_POWER_INCLUSION_LOWER_BOUND class-attribute instance-attribute ¤
ACTIVE_POWER_INCLUSION_LOWER_BOUND = (
    "active_power_inclusion_lower_bound"
)

Active power inclusion lower bound.

ACTIVE_POWER_INCLUSION_UPPER_BOUND class-attribute instance-attribute ¤
ACTIVE_POWER_INCLUSION_UPPER_BOUND = (
    "active_power_inclusion_upper_bound"
)

Active power inclusion upper bound.

ACTIVE_POWER_PHASE_1 class-attribute instance-attribute ¤
ACTIVE_POWER_PHASE_1 = 'active_power_phase_1'

Active power in phase 1.

ACTIVE_POWER_PHASE_2 class-attribute instance-attribute ¤
ACTIVE_POWER_PHASE_2 = 'active_power_phase_2'

Active power in phase 2.

ACTIVE_POWER_PHASE_3 class-attribute instance-attribute ¤
ACTIVE_POWER_PHASE_3 = 'active_power_phase_3'

Active power in phase 3.

CAPACITY class-attribute instance-attribute ¤
CAPACITY = 'capacity'

Capacity.

CURRENT_PHASE_1 class-attribute instance-attribute ¤
CURRENT_PHASE_1 = 'current_phase_1'

Current in phase 1.

CURRENT_PHASE_2 class-attribute instance-attribute ¤
CURRENT_PHASE_2 = 'current_phase_2'

Current in phase 2.

CURRENT_PHASE_3 class-attribute instance-attribute ¤
CURRENT_PHASE_3 = 'current_phase_3'

Current in phase 3.

POWER_EXCLUSION_LOWER_BOUND class-attribute instance-attribute ¤
POWER_EXCLUSION_LOWER_BOUND = 'power_exclusion_lower_bound'

Power exclusion lower bound.

POWER_EXCLUSION_UPPER_BOUND class-attribute instance-attribute ¤
POWER_EXCLUSION_UPPER_BOUND = 'power_exclusion_upper_bound'

Power exclusion upper bound.

POWER_INCLUSION_LOWER_BOUND class-attribute instance-attribute ¤
POWER_INCLUSION_LOWER_BOUND = 'power_inclusion_lower_bound'

Power inclusion lower bound.

POWER_INCLUSION_UPPER_BOUND class-attribute instance-attribute ¤
POWER_INCLUSION_UPPER_BOUND = 'power_inclusion_upper_bound'

Power inclusion upper bound.

REACTIVE_POWER class-attribute instance-attribute ¤
REACTIVE_POWER = 'reactive_power'

Reactive power.

REACTIVE_POWER_PHASE_1 class-attribute instance-attribute ¤
REACTIVE_POWER_PHASE_1 = 'reactive_power_phase_1'

Reactive power in phase 1.

REACTIVE_POWER_PHASE_2 class-attribute instance-attribute ¤
REACTIVE_POWER_PHASE_2 = 'reactive_power_phase_2'

Reactive power in phase 2.

REACTIVE_POWER_PHASE_3 class-attribute instance-attribute ¤
REACTIVE_POWER_PHASE_3 = 'reactive_power_phase_3'

Reactive power in phase 3.

SOC class-attribute instance-attribute ¤
SOC = 'soc'

State of charge.

SOC_LOWER_BOUND class-attribute instance-attribute ¤
SOC_LOWER_BOUND = 'soc_lower_bound'

Lower bound of state of charge.

SOC_UPPER_BOUND class-attribute instance-attribute ¤
SOC_UPPER_BOUND = 'soc_upper_bound'

Upper bound of state of charge.

TEMPERATURE class-attribute instance-attribute ¤
TEMPERATURE = 'temperature'

Temperature.

VOLTAGE_PHASE_1 class-attribute instance-attribute ¤
VOLTAGE_PHASE_1 = 'voltage_phase_1'

Voltage in phase 1.

VOLTAGE_PHASE_2 class-attribute instance-attribute ¤
VOLTAGE_PHASE_2 = 'voltage_phase_2'

Voltage in phase 2.

VOLTAGE_PHASE_3 class-attribute instance-attribute ¤
VOLTAGE_PHASE_3 = 'voltage_phase_3'

Voltage in phase 3.

frequenz.client.microgrid.ComponentType ¤

Bases: Enum

A base class from which individual component types are derived.

Source code in frequenz/client/microgrid/_component.py
class ComponentType(Enum):
    """A base class from which individual component types are derived."""

frequenz.client.microgrid.Connection dataclass ¤

Metadata for a connection between microgrid components.

Source code in frequenz/client/microgrid/_connection.py
@dataclass(frozen=True)
class Connection:
    """Metadata for a connection between microgrid components."""

    start: int
    """The component ID that represents the start component of the connection."""

    end: int
    """The component ID that represents the end component of the connection."""

    def is_valid(self) -> bool:
        """Check if this instance contains valid data.

        Returns:
            `True` if `start >= 0`, `end > 0`, and `start != end`, `False`
                otherwise.
        """
        return self.start >= 0 and self.end > 0 and self.start != self.end
Attributes¤
end instance-attribute ¤
end: int

The component ID that represents the end component of the connection.

start instance-attribute ¤
start: int

The component ID that represents the start component of the connection.

Functions¤
is_valid ¤
is_valid() -> bool

Check if this instance contains valid data.

RETURNS DESCRIPTION
bool

True if start >= 0, end > 0, and start != end, False otherwise.

Source code in frequenz/client/microgrid/_connection.py
def is_valid(self) -> bool:
    """Check if this instance contains valid data.

    Returns:
        `True` if `start >= 0`, `end > 0`, and `start != end`, `False`
            otherwise.
    """
    return self.start >= 0 and self.end > 0 and self.start != self.end

frequenz.client.microgrid.DataLoss ¤

Bases: GrpcError

Unrecoverable data loss or corruption.

Source code in frequenz/client/base/exception.py
class DataLoss(GrpcError):
    """Unrecoverable data loss or corruption."""

    def __init__(
        self, *, server_url: str, operation: str, grpc_error: AioRpcError
    ) -> None:
        """Create a new instance.

        Args:
            server_url: The URL of the server that returned the error.
            operation: The operation that caused the error.
            grpc_error: The gRPC error originating this exception.
        """
        super().__init__(
            server_url=server_url,
            operation=operation,
            description="Unrecoverable data loss or corruption",
            grpc_error=grpc_error,
            retryable=False,
        )
Attributes¤
description instance-attribute ¤
description: str = description

The human-readable description of the error.

grpc_error instance-attribute ¤
grpc_error: AioRpcError = grpc_error

The original gRPC error.

is_retryable instance-attribute ¤
is_retryable = retryable

Whether retrying the operation might succeed.

operation instance-attribute ¤
operation = operation

The operation that caused the error.

server_url instance-attribute ¤
server_url = server_url

The URL of the server that returned the error.

Functions¤
__init__ ¤
__init__(
    *,
    server_url: str,
    operation: str,
    grpc_error: AioRpcError
) -> None

Create a new instance.

PARAMETER DESCRIPTION
server_url

The URL of the server that returned the error.

TYPE: str

operation

The operation that caused the error.

TYPE: str

grpc_error

The gRPC error originating this exception.

TYPE: AioRpcError

Source code in frequenz/client/base/exception.py
def __init__(
    self, *, server_url: str, operation: str, grpc_error: AioRpcError
) -> None:
    """Create a new instance.

    Args:
        server_url: The URL of the server that returned the error.
        operation: The operation that caused the error.
        grpc_error: The gRPC error originating this exception.
    """
    super().__init__(
        server_url=server_url,
        operation=operation,
        description="Unrecoverable data loss or corruption",
        grpc_error=grpc_error,
        retryable=False,
    )
from_grpc_error classmethod ¤
from_grpc_error(
    *,
    server_url: str,
    operation: str,
    grpc_error: AioRpcError
) -> GrpcError

Create an instance of the appropriate subclass from a gRPC error.

PARAMETER DESCRIPTION
server_url

The URL of the server that returned the error.

TYPE: str

operation

The operation that caused the error.

TYPE: str

grpc_error

The gRPC error to convert.

TYPE: AioRpcError

RETURNS DESCRIPTION
GrpcError

An instance of GrpcError if the gRPC status is not recognized, or an appropriate subclass if it is.

Source code in frequenz/client/base/exception.py
@classmethod
def from_grpc_error(
    cls,
    *,
    server_url: str,
    operation: str,
    grpc_error: AioRpcError,
) -> GrpcError:
    """Create an instance of the appropriate subclass from a gRPC error.

    Args:
        server_url: The URL of the server that returned the error.
        operation: The operation that caused the error.
        grpc_error: The gRPC error to convert.

    Returns:
        An instance of
            [GrpcError][frequenz.client.base.exception.GrpcError] if the gRPC status
            is not recognized, or an appropriate subclass if it is.
    """

    class Ctor(Protocol):
        """A protocol for the constructor of a subclass of `GrpcError`."""

        def __call__(
            self, *, server_url: str, operation: str, grpc_error: AioRpcError
        ) -> GrpcError: ...

    grpc_status_map: dict[grpc.StatusCode, Ctor] = {
        grpc.StatusCode.CANCELLED: OperationCancelled,
        grpc.StatusCode.UNKNOWN: UnknownError,
        grpc.StatusCode.INVALID_ARGUMENT: InvalidArgument,
        grpc.StatusCode.DEADLINE_EXCEEDED: OperationTimedOut,
        grpc.StatusCode.NOT_FOUND: EntityNotFound,
        grpc.StatusCode.ALREADY_EXISTS: EntityAlreadyExists,
        grpc.StatusCode.PERMISSION_DENIED: PermissionDenied,
        grpc.StatusCode.RESOURCE_EXHAUSTED: ResourceExhausted,
        grpc.StatusCode.FAILED_PRECONDITION: OperationPreconditionFailed,
        grpc.StatusCode.ABORTED: OperationAborted,
        grpc.StatusCode.OUT_OF_RANGE: OperationOutOfRange,
        grpc.StatusCode.UNIMPLEMENTED: OperationNotImplemented,
        grpc.StatusCode.INTERNAL: InternalError,
        grpc.StatusCode.UNAVAILABLE: ServiceUnavailable,
        grpc.StatusCode.DATA_LOSS: DataLoss,
        grpc.StatusCode.UNAUTHENTICATED: OperationUnauthenticated,
    }

    if ctor := grpc_status_map.get(grpc_error.code()):
        return ctor(
            server_url=server_url, operation=operation, grpc_error=grpc_error
        )
    return UnrecognizedGrpcStatus(
        server_url=server_url,
        operation=operation,
        grpc_error=grpc_error,
    )

frequenz.client.microgrid.EVChargerCableState ¤

Bases: Enum

Cable states of an EV Charger.

Source code in frequenz/client/microgrid/_component_states.py
class EVChargerCableState(Enum):
    """Cable states of an EV Charger."""

    UNSPECIFIED = ev_charger_pb2.CableState.CABLE_STATE_UNSPECIFIED
    """Unspecified cable state."""

    UNPLUGGED = ev_charger_pb2.CableState.CABLE_STATE_UNPLUGGED
    """The cable is unplugged."""

    CHARGING_STATION_PLUGGED = (
        ev_charger_pb2.CableState.CABLE_STATE_CHARGING_STATION_PLUGGED
    )
    """The cable is plugged into the charging station."""

    CHARGING_STATION_LOCKED = (
        ev_charger_pb2.CableState.CABLE_STATE_CHARGING_STATION_LOCKED
    )
    """The cable is plugged into the charging station and locked."""

    EV_PLUGGED = ev_charger_pb2.CableState.CABLE_STATE_EV_PLUGGED
    """The cable is plugged into the EV."""

    EV_LOCKED = ev_charger_pb2.CableState.CABLE_STATE_EV_LOCKED
    """The cable is plugged into the EV and locked."""

    @classmethod
    def from_pb(cls, state: ev_charger_pb2.CableState.ValueType) -> Self:
        """Convert a protobuf state value to this enum.

        Args:
            state: The protobuf cable state to convert.

        Returns:
            The enum value corresponding to the protobuf message.
        """
        try:
            return cls(state)
        except ValueError:
            return cls(cls.UNSPECIFIED)
Attributes¤
CHARGING_STATION_LOCKED class-attribute instance-attribute ¤
CHARGING_STATION_LOCKED = (
    CABLE_STATE_CHARGING_STATION_LOCKED
)

The cable is plugged into the charging station and locked.

CHARGING_STATION_PLUGGED class-attribute instance-attribute ¤
CHARGING_STATION_PLUGGED = (
    CABLE_STATE_CHARGING_STATION_PLUGGED
)

The cable is plugged into the charging station.

EV_LOCKED class-attribute instance-attribute ¤
EV_LOCKED = CABLE_STATE_EV_LOCKED

The cable is plugged into the EV and locked.

EV_PLUGGED class-attribute instance-attribute ¤
EV_PLUGGED = CABLE_STATE_EV_PLUGGED

The cable is plugged into the EV.

UNPLUGGED class-attribute instance-attribute ¤
UNPLUGGED = CABLE_STATE_UNPLUGGED

The cable is unplugged.

UNSPECIFIED class-attribute instance-attribute ¤
UNSPECIFIED = CABLE_STATE_UNSPECIFIED

Unspecified cable state.

Functions¤
from_pb classmethod ¤
from_pb(state: ValueType) -> Self

Convert a protobuf state value to this enum.

PARAMETER DESCRIPTION
state

The protobuf cable state to convert.

TYPE: ValueType

RETURNS DESCRIPTION
Self

The enum value corresponding to the protobuf message.

Source code in frequenz/client/microgrid/_component_states.py
@classmethod
def from_pb(cls, state: ev_charger_pb2.CableState.ValueType) -> Self:
    """Convert a protobuf state value to this enum.

    Args:
        state: The protobuf cable state to convert.

    Returns:
        The enum value corresponding to the protobuf message.
    """
    try:
        return cls(state)
    except ValueError:
        return cls(cls.UNSPECIFIED)

frequenz.client.microgrid.EVChargerComponentState ¤

Bases: Enum

Component State of an EV Charger.

Source code in frequenz/client/microgrid/_component_states.py
class EVChargerComponentState(Enum):
    """Component State of an EV Charger."""

    UNSPECIFIED = ev_charger_pb2.ComponentState.COMPONENT_STATE_UNSPECIFIED
    """Unspecified component state."""

    STARTING = ev_charger_pb2.ComponentState.COMPONENT_STATE_STARTING
    """The component is starting."""

    NOT_READY = ev_charger_pb2.ComponentState.COMPONENT_STATE_NOT_READY
    """The component is not ready."""

    READY = ev_charger_pb2.ComponentState.COMPONENT_STATE_READY
    """The component is ready."""

    CHARGING = ev_charger_pb2.ComponentState.COMPONENT_STATE_CHARGING
    """The component is charging."""

    DISCHARGING = ev_charger_pb2.ComponentState.COMPONENT_STATE_DISCHARGING
    """The component is discharging."""

    ERROR = ev_charger_pb2.ComponentState.COMPONENT_STATE_ERROR
    """The component is in error state."""

    AUTHORIZATION_REJECTED = (
        ev_charger_pb2.ComponentState.COMPONENT_STATE_AUTHORIZATION_REJECTED
    )
    """The component rejected authorization."""

    INTERRUPTED = ev_charger_pb2.ComponentState.COMPONENT_STATE_INTERRUPTED
    """The component is interrupted."""

    UNKNOWN = ev_charger_pb2.ComponentState.COMPONENT_STATE_UNKNOWN
    """A state is provided by the component, but it is not one of the above states."""

    @classmethod
    def from_pb(cls, state: ev_charger_pb2.ComponentState.ValueType) -> Self:
        """Convert a protobuf state value to this enum.

        Args:
            state: The protobuf component state to convert.

        Returns:
            The enum value corresponding to the protobuf message.
        """
        try:
            return cls(state)
        except ValueError:
            return cls(cls.UNKNOWN)
Attributes¤
AUTHORIZATION_REJECTED class-attribute instance-attribute ¤
AUTHORIZATION_REJECTED = (
    COMPONENT_STATE_AUTHORIZATION_REJECTED
)

The component rejected authorization.

CHARGING class-attribute instance-attribute ¤
CHARGING = COMPONENT_STATE_CHARGING

The component is charging.

DISCHARGING class-attribute instance-attribute ¤
DISCHARGING = COMPONENT_STATE_DISCHARGING

The component is discharging.

ERROR class-attribute instance-attribute ¤
ERROR = COMPONENT_STATE_ERROR

The component is in error state.

INTERRUPTED class-attribute instance-attribute ¤
INTERRUPTED = COMPONENT_STATE_INTERRUPTED

The component is interrupted.

NOT_READY class-attribute instance-attribute ¤
NOT_READY = COMPONENT_STATE_NOT_READY

The component is not ready.

READY class-attribute instance-attribute ¤
READY = COMPONENT_STATE_READY

The component is ready.

STARTING class-attribute instance-attribute ¤
STARTING = COMPONENT_STATE_STARTING

The component is starting.

UNKNOWN class-attribute instance-attribute ¤
UNKNOWN = COMPONENT_STATE_UNKNOWN

A state is provided by the component, but it is not one of the above states.

UNSPECIFIED class-attribute instance-attribute ¤
UNSPECIFIED = COMPONENT_STATE_UNSPECIFIED

Unspecified component state.

Functions¤
from_pb classmethod ¤
from_pb(state: ValueType) -> Self

Convert a protobuf state value to this enum.

PARAMETER DESCRIPTION
state

The protobuf component state to convert.

TYPE: ValueType

RETURNS DESCRIPTION
Self

The enum value corresponding to the protobuf message.

Source code in frequenz/client/microgrid/_component_states.py
@classmethod
def from_pb(cls, state: ev_charger_pb2.ComponentState.ValueType) -> Self:
    """Convert a protobuf state value to this enum.

    Args:
        state: The protobuf component state to convert.

    Returns:
        The enum value corresponding to the protobuf message.
    """
    try:
        return cls(state)
    except ValueError:
        return cls(cls.UNKNOWN)

frequenz.client.microgrid.EVChargerData dataclass ¤

Bases: ComponentData

A wrapper class for holding ev_charger data.

Source code in frequenz/client/microgrid/_component_data.py
@dataclass(frozen=True)
class EVChargerData(ComponentData):  # pylint: disable=too-many-instance-attributes
    """A wrapper class for holding ev_charger data."""

    active_power: float
    """The total active 3-phase AC power, in Watts (W).

    Represented in the passive sign convention.

    * Positive means consumption from the grid.
    * Negative means supply into the grid.
    """

    active_power_per_phase: tuple[float, float, float]
    """The per-phase AC active power for phase 1, 2, and 3 respectively, in Watt (W).

    Represented in the passive sign convention.

    * Positive means consumption from the grid.
    * Negative means supply into the grid.
    """

    current_per_phase: tuple[float, float, float]
    """AC current in Amperes (A) for phase/line 1,2 and 3 respectively.

    Represented in the passive sign convention.

    * Positive means consumption from the grid.
    * Negative means supply into the grid.
    """

    reactive_power: float
    """The total reactive 3-phase AC power, in Volt-Ampere Reactive (VAr).

    * Positive power means capacitive (current leading w.r.t. voltage).
    * Negative power means inductive (current lagging w.r.t. voltage).
    """

    reactive_power_per_phase: tuple[float, float, float]
    """The per-phase AC reactive power, in Volt-Ampere Reactive (VAr).

    The provided values are for phase 1, 2, and 3 respectively.

    * Positive power means capacitive (current leading w.r.t. voltage).
    * Negative power means inductive (current lagging w.r.t. voltage).
    """

    voltage_per_phase: tuple[float, float, float]
    """The AC voltage in Volts (V) between the line and the neutral
        wire for phase/line 1,2 and 3 respectively.
    """

    active_power_inclusion_lower_bound: float
    """Lower inclusion bound for EV charger power in watts.

    This is the lower limit of the range within which power requests are allowed for the
    EV charger.

    See [`frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds`][] and
    [`frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds`][] for more
    details.
    """

    active_power_exclusion_lower_bound: float
    """Lower exclusion bound for EV charger power in watts.

    This is the lower limit of the range within which power requests are not allowed for
    the EV charger.

    See [`frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds`][] and
    [`frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds`][] for more
    details.
    """

    active_power_inclusion_upper_bound: float
    """Upper inclusion bound for EV charger power in watts.

    This is the upper limit of the range within which power requests are allowed for the
    EV charger.

    See [`frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds`][] and
    [`frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds`][] for more
    details.
    """

    active_power_exclusion_upper_bound: float
    """Upper exclusion bound for EV charger power in watts.

    This is the upper limit of the range within which power requests are not allowed for
    the EV charger.

    See [`frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds`][] and
    [`frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds`][] for more
    details.
    """

    frequency: float
    """AC frequency, in Hertz (Hz)."""

    cable_state: EVChargerCableState
    """The state of the ev charger's cable."""

    component_state: EVChargerComponentState
    """The state of the ev charger."""

    @classmethod
    def from_proto(cls, raw: microgrid_pb2.ComponentData) -> Self:
        """Create EVChargerData from a protobuf message.

        Args:
            raw: raw component data as decoded from the wire.

        Returns:
            Instance of EVChargerData created from the protobuf message.
        """
        raw_power = raw.ev_charger.data.ac.power_active
        ev_charger_data = cls(
            component_id=raw.id,
            timestamp=raw.ts.ToDatetime(tzinfo=timezone.utc),
            active_power=raw_power.value,
            active_power_per_phase=(
                raw.ev_charger.data.ac.phase_1.power_active.value,
                raw.ev_charger.data.ac.phase_2.power_active.value,
                raw.ev_charger.data.ac.phase_3.power_active.value,
            ),
            reactive_power=raw.ev_charger.data.ac.power_reactive.value,
            reactive_power_per_phase=(
                raw.ev_charger.data.ac.phase_1.power_reactive.value,
                raw.ev_charger.data.ac.phase_2.power_reactive.value,
                raw.ev_charger.data.ac.phase_3.power_reactive.value,
            ),
            current_per_phase=(
                raw.ev_charger.data.ac.phase_1.current.value,
                raw.ev_charger.data.ac.phase_2.current.value,
                raw.ev_charger.data.ac.phase_3.current.value,
            ),
            voltage_per_phase=(
                raw.ev_charger.data.ac.phase_1.voltage.value,
                raw.ev_charger.data.ac.phase_2.voltage.value,
                raw.ev_charger.data.ac.phase_3.voltage.value,
            ),
            active_power_inclusion_lower_bound=raw_power.system_inclusion_bounds.lower,
            active_power_exclusion_lower_bound=raw_power.system_exclusion_bounds.lower,
            active_power_inclusion_upper_bound=raw_power.system_inclusion_bounds.upper,
            active_power_exclusion_upper_bound=raw_power.system_exclusion_bounds.upper,
            cable_state=EVChargerCableState.from_pb(raw.ev_charger.state.cable_state),
            component_state=EVChargerComponentState.from_pb(
                raw.ev_charger.state.component_state
            ),
            frequency=raw.ev_charger.data.ac.frequency.value,
        )
        ev_charger_data._set_raw(raw=raw)
        return ev_charger_data

    def is_ev_connected(self) -> bool:
        """Check whether an EV is connected to the charger.

        Returns:
            When the charger is not in an error state, whether an EV is connected to
                the charger.
        """
        return self.component_state not in (
            EVChargerComponentState.AUTHORIZATION_REJECTED,
            EVChargerComponentState.ERROR,
        ) and self.cable_state in (
            EVChargerCableState.EV_LOCKED,
            EVChargerCableState.EV_PLUGGED,
        )
Attributes¤
active_power instance-attribute ¤
active_power: float

The total active 3-phase AC power, in Watts (W).

Represented in the passive sign convention.

  • Positive means consumption from the grid.
  • Negative means supply into the grid.
active_power_exclusion_lower_bound instance-attribute ¤
active_power_exclusion_lower_bound: float

Lower exclusion bound for EV charger power in watts.

This is the lower limit of the range within which power requests are not allowed for the EV charger.

See frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds and frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds for more details.

active_power_exclusion_upper_bound instance-attribute ¤
active_power_exclusion_upper_bound: float

Upper exclusion bound for EV charger power in watts.

This is the upper limit of the range within which power requests are not allowed for the EV charger.

See frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds and frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds for more details.

active_power_inclusion_lower_bound instance-attribute ¤
active_power_inclusion_lower_bound: float

Lower inclusion bound for EV charger power in watts.

This is the lower limit of the range within which power requests are allowed for the EV charger.

See frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds and frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds for more details.

active_power_inclusion_upper_bound instance-attribute ¤
active_power_inclusion_upper_bound: float

Upper inclusion bound for EV charger power in watts.

This is the upper limit of the range within which power requests are allowed for the EV charger.

See frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds and frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds for more details.

active_power_per_phase instance-attribute ¤
active_power_per_phase: tuple[float, float, float]

The per-phase AC active power for phase 1, 2, and 3 respectively, in Watt (W).

Represented in the passive sign convention.

  • Positive means consumption from the grid.
  • Negative means supply into the grid.
cable_state instance-attribute ¤
cable_state: EVChargerCableState

The state of the ev charger's cable.

component_id instance-attribute ¤
component_id: int

The ID identifying this component in the microgrid.

component_state instance-attribute ¤
component_state: EVChargerComponentState

The state of the ev charger.

current_per_phase instance-attribute ¤
current_per_phase: tuple[float, float, float]

AC current in Amperes (A) for phase/line 1,2 and 3 respectively.

Represented in the passive sign convention.

  • Positive means consumption from the grid.
  • Negative means supply into the grid.
frequency instance-attribute ¤
frequency: float

AC frequency, in Hertz (Hz).

raw class-attribute instance-attribute ¤
raw: ComponentData | None = field(default=None, init=False)

Raw component data as decoded from the wire.

reactive_power instance-attribute ¤
reactive_power: