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

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:

GrpcError: A gRPC operation failed.

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` ¤

code: BatteryErrorCode = UNSPECIFIED

The error code.

level `class-attribute` `instance-attribute` ¤

level: ErrorLevel = UNSPECIFIED

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` ¤

category: ComponentCategory

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: 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 `instance-attribute` ¤

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).

timestamp `instance-attribute` ¤

timestamp: datetime

The timestamp of when the data was measured.

voltage_per_phase `instance-attribute` ¤

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.

Functions¤

from_proto `classmethod` ¤

from_proto(raw: ComponentData) -> Self

Create EVChargerData from a protobuf message.

PARAMETER	DESCRIPTION
`raw`	raw component data as decoded from the wire. TYPE: `ComponentData`

RETURNS	DESCRIPTION
`Self`	Instance of EVChargerData 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 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

is_ev_connected ¤

is_ev_connected() -> bool

Check whether an EV is connected to the charger.

RETURNS	DESCRIPTION
`bool`	When the charger is not in an error state, whether an EV is connected to the charger.

Source code in frequenz/client/microgrid/_component_data.py

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,
    )

frequenz.client.microgrid.EntityAlreadyExists ¤

Bases: GrpcError

The entity that we attempted to create already exists.

Source code in frequenz/client/base/exception.py

class EntityAlreadyExists(GrpcError):
    """The entity that we attempted to create already exists."""

    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="The entity that we attempted to create already exists",
            grpc_error=grpc_error,
            retryable=True,  # If the entity is deleted later it might succeed
        )

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="The entity that we attempted to create already exists",
        grpc_error=grpc_error,
        retryable=True,  # If the entity is deleted later it 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.EntityNotFound ¤

Bases: GrpcError

The requested entity was not found.

Note that this error differs from PermissionDenied. This error is used when the requested entity is not found, regardless of the user's permissions.

Source code in frequenz/client/base/exception.py

class EntityNotFound(GrpcError):
    """The requested entity was not found.

    Note that this error differs from
    [PermissionDenied][frequenz.client.base.exception.PermissionDenied]. This error is
    used when the requested entity is not found, regardless of the user's permissions.
    """

    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="The requested entity was not found",
            grpc_error=grpc_error,
            retryable=True,  # If the entity is added later it might succeed
        )

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="The requested entity was not found",
        grpc_error=grpc_error,
        retryable=True,  # If the entity is added later it 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.ErrorLevel ¤

Bases: Enum

Error level.

Source code in frequenz/client/microgrid/_component_error.py

class ErrorLevel(Enum):
    """Error level."""

    UNSPECIFIED = common_pb2.ErrorLevel.ERROR_LEVEL_UNSPECIFIED
    """Unspecified component error."""

    WARN = common_pb2.ErrorLevel.ERROR_LEVEL_WARN
    """Action must be taken to prevent a severe error from occurring in the future."""

    CRITICAL = common_pb2.ErrorLevel.ERROR_LEVEL_CRITICAL
    """A severe error that causes the component to fail. Immediate action must be taken."""

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

        Args:
            code: The protobuf error level to convert.

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

Attributes¤

CRITICAL `class-attribute` `instance-attribute` ¤

CRITICAL = ERROR_LEVEL_CRITICAL

A severe error that causes the component to fail. Immediate action must be taken.

UNSPECIFIED `class-attribute` `instance-attribute` ¤

UNSPECIFIED = ERROR_LEVEL_UNSPECIFIED

Unspecified component error.

WARN `class-attribute` `instance-attribute` ¤

WARN = ERROR_LEVEL_WARN

Action must be taken to prevent a severe error from occurring in the future.

Functions¤

from_pb `classmethod` ¤

from_pb(code: ValueType) -> Self

Convert a protobuf error level value to this enum.

PARAMETER	DESCRIPTION
`code`	The protobuf error level 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: common_pb2.ErrorLevel.ValueType) -> Self:
    """Convert a protobuf error level value to this enum.

    Args:
        code: The protobuf error level to convert.

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

frequenz.client.microgrid.Fuse `dataclass` ¤

Fuse data class.

Source code in frequenz/client/microgrid/_component.py

@dataclass(frozen=True)
class Fuse:
    """Fuse data class."""

    max_current: float
    """Rated current of the fuse."""

Attributes¤

max_current `instance-attribute` ¤

max_current: float

Rated current of the fuse.

frequenz.client.microgrid.GridMetadata `dataclass` ¤

Bases: ComponentMetadata

Metadata for a grid connection point.

Source code in frequenz/client/microgrid/_component.py

@dataclass(frozen=True)
class GridMetadata(ComponentMetadata):
    """Metadata for a grid connection point."""

Attributes¤

fuse `class-attribute` `instance-attribute` ¤

fuse: Fuse | None = None

The fuse at the grid connection point.

frequenz.client.microgrid.GrpcError ¤

Bases: ApiClientError

The gRPC server returned an error with a status code.

These errors are specific to gRPC. If you want to use the client in a protocol-independent way, you should avoid catching this exception. Catching subclasses that don't have grpc in their name should be protocol-independent.

The following sub-classes are available:

DataLoss: Unrecoverable data loss or corruption.
EntityAlreadyExists: The entity that we attempted to create already exists.
EntityNotFound: The requested entity was not found.
InternalError: Some invariants expected by the underlying system have been broken.
InvalidArgument: The client specified an invalid argument.
OperationAborted: The operation was aborted.
OperationCancelled: The operation was cancelled.
OperationNotImplemented: The operation is not implemented or not supported/enabled in this service.
OperationOutOfRange: The operation was attempted past the valid range.
OperationPreconditionFailed: The operation was rejected because the system is not in a required state.
OperationTimedOut: The time limit was exceeded while waiting for the operation to complete.
OperationUnauthenticated: The request does not have valid authentication credentials for the operation.
PermissionDenied: The caller does not have permission to execute the specified operation.
ResourceExhausted: Some resource has been exhausted (for example per-user quota, disk space, etc.).
ServiceUnavailable: The service is currently unavailable.
UnknownError: There was an error that can't be described using other statuses.
UnrecognizedGrpcStatus: The gRPC server returned an unrecognized status code.

References

gRPC status codes

Source code in frequenz/client/base/exception.py

class GrpcError(ApiClientError):
    """The gRPC server returned an error with a status code.

    These errors are specific to gRPC. If you want to use the client in
    a protocol-independent way, you should avoid catching this exception. Catching
    subclasses that don't have *grpc* in their name should be protocol-independent.

    The following sub-classes are available:

    - [DataLoss][frequenz.client.base.exception.DataLoss]: Unrecoverable data loss or
      corruption.
    - [EntityAlreadyExists][frequenz.client.base.exception.EntityAlreadyExists]: The
      entity that we attempted to create already exists.
    - [EntityNotFound][frequenz.client.base.exception.EntityNotFound]: The requested
      entity was not found.
    - [InternalError][frequenz.client.base.exception.InternalError]: Some invariants
      expected by the underlying system have been broken.
    - [InvalidArgument][frequenz.client.base.exception.InvalidArgument]: The client
      specified an invalid argument.
    - [OperationAborted][frequenz.client.base.exception.OperationAborted]: The
      operation was aborted.
    - [OperationCancelled][frequenz.client.base.exception.OperationCancelled]: The
      operation was cancelled.
    - [OperationNotImplemented][frequenz.client.base.exception.OperationNotImplemented]:
      The operation is not implemented or not supported/enabled in this service.
    - [OperationOutOfRange][frequenz.client.base.exception.OperationOutOfRange]: The
      operation was attempted past the valid range.
    - [OperationPreconditionFailed][frequenz.client.base.exception.OperationPreconditionFailed]:
      The operation was rejected because the system is not in a required state.
    - [OperationTimedOut][frequenz.client.base.exception.OperationTimedOut]: The time
      limit was exceeded while waiting for the operation to complete.
    - [OperationUnauthenticated][frequenz.client.base.exception.OperationUnauthenticated]:
      The request does not have valid authentication credentials for the operation.
    - [PermissionDenied][frequenz.client.base.exception.PermissionDenied]: The caller
      does not have permission to execute the specified operation.
    - [ResourceExhausted][frequenz.client.base.exception.ResourceExhausted]: Some
      resource has been exhausted (for example per-user quota, disk space, etc.).
    - [ServiceUnavailable][frequenz.client.base.exception.ServiceUnavailable]: The
      service is currently unavailable.
    - [UnknownError][frequenz.client.base.exception.UnknownError]: There was an error
      that can't be described using other statuses.
    - [UnrecognizedGrpcStatus][frequenz.client.base.exception.UnrecognizedGrpcStatus]:
      The gRPC server returned an unrecognized status code.

    References:
        * [gRPC status
           codes](https://github.com/grpc/grpc/blob/master/doc/statuscodes.md)
    """

    def __init__(  # pylint: disable=too-many-arguments
        self,
        *,
        server_url: str,
        operation: str,
        description: str,
        grpc_error: AioRpcError,
        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.
            grpc_error: The gRPC error originating this exception.
            retryable: Whether retrying the operation might succeed.
        """
        status_name = grpc_error.code().name
        message = grpc_error.details()
        details = grpc_error.debug_error_string()
        message = f": {message}" if message else ""
        details = f" ({details})" if details else ""
        super().__init__(
            server_url=server_url,
            operation=operation,
            description=f"{description} <status={status_name}>{message}{details}",
            retryable=retryable,
        )
        self.description: str = description
        """The human-readable description of the error."""

        self.grpc_error: AioRpcError = grpc_error
        """The original gRPC error."""

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,
    description: str,
    grpc_error: AioRpcError,
    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`
`grpc_error`	The gRPC error originating this exception. TYPE: `AioRpcError`
`retryable`	Whether retrying the operation might succeed. TYPE: `bool`

Source code in frequenz/client/base/exception.py

def __init__(  # pylint: disable=too-many-arguments
    self,
    *,
    server_url: str,
    operation: str,
    description: str,
    grpc_error: AioRpcError,
    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.
        grpc_error: The gRPC error originating this exception.
        retryable: Whether retrying the operation might succeed.
    """
    status_name = grpc_error.code().name
    message = grpc_error.details()
    details = grpc_error.debug_error_string()
    message = f": {message}" if message else ""
    details = f" ({details})" if details else ""
    super().__init__(
        server_url=server_url,
        operation=operation,
        description=f"{description} <status={status_name}>{message}{details}",
        retryable=retryable,
    )
    self.description: str = description
    """The human-readable description of the error."""

    self.grpc_error: AioRpcError = grpc_error
    """The original gRPC error."""

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.InternalError ¤

Bases: GrpcError

Some invariants expected by the underlying system have been broken.

This error code is reserved for serious errors.

Source code in frequenz/client/base/exception.py

class InternalError(GrpcError):
    """Some invariants expected by the underlying system have been broken.

    This error code is reserved for serious errors.
    """

    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="Some invariants expected by the underlying system have been "
            "broken",
            grpc_error=grpc_error,
            retryable=True,  # If the system state changes it might succeed
        )

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="Some invariants expected by the underlying system have been "
        "broken",
        grpc_error=grpc_error,
        retryable=True,  # If the system state changes it 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.InvalidArgument ¤

Bases: GrpcError, ValueError

The client specified an invalid argument.

Note that this error differs from OperationPreconditionFailed. This error indicates arguments that are problematic regardless of the state of the system (e.g., a malformed file name).

Source code in frequenz/client/base/exception.py

class InvalidArgument(GrpcError, ValueError):
    """The client specified an invalid argument.

    Note that this error differs from
    [OperationPreconditionFailed][frequenz.client.base.exception.OperationPreconditionFailed].
    This error indicates arguments that are problematic regardless of the state of the
    system (e.g., a malformed file name).
    """

    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="The client specified an invalid argument",
            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="The client specified an invalid argument",
        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.InverterComponentState ¤

Bases: Enum

Component states of an inverter.

Source code in frequenz/client/microgrid/_component_states.py

class InverterComponentState(Enum):
    """Component states of an inverter."""

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

    OFF = inverter_pb2.ComponentState.COMPONENT_STATE_OFF
    """Inverter is switched off."""

    SWITCHING_ON = inverter_pb2.ComponentState.COMPONENT_STATE_SWITCHING_ON
    """The PbInverteris starting up and needs some time to become fully operational."""

    SWITCHING_OFF = inverter_pb2.ComponentState.COMPONENT_STATE_SWITCHING_OFF
    """The PbInverteris switching off and needs some time to fully shut down."""

    STANDBY = inverter_pb2.ComponentState.COMPONENT_STATE_STANDBY
    """The PbInverteris in a standby state, and is disconnected from the grid.

    When connected to the grid, it run a few tests, and move to the `IDLE` state.
    """

    IDLE = inverter_pb2.ComponentState.COMPONENT_STATE_IDLE
    """The inverter is idle."""

    CHARGING = inverter_pb2.ComponentState.COMPONENT_STATE_CHARGING
    """The inverter is consuming electrical energy to charge batteries.

    Applicable to `BATTERY` and `HYBRID` inverters only.
    """

    DISCHARGING = inverter_pb2.ComponentState.COMPONENT_STATE_DISCHARGING
    """The inverter is generating electrical energy."""

    ERROR = inverter_pb2.ComponentState.COMPONENT_STATE_ERROR
    """The inverter is in a faulty state."""

    UNAVAILABLE = inverter_pb2.ComponentState.COMPONENT_STATE_UNAVAILABLE
    """The inverter is online, but currently unavailable.

    Possibly due to a pre- scheduled maintenance.
    """

    UNKNOWN = inverter_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: inverter_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 inverter is consuming electrical energy to charge batteries.

Applicable to BATTERY and HYBRID inverters only.

DISCHARGING `class-attribute` `instance-attribute` ¤

DISCHARGING = COMPONENT_STATE_DISCHARGING

The inverter is generating electrical energy.

ERROR `class-attribute` `instance-attribute` ¤

ERROR = COMPONENT_STATE_ERROR

The inverter is in a faulty state.

IDLE `class-attribute` `instance-attribute` ¤

IDLE = COMPONENT_STATE_IDLE

The inverter is idle.

OFF `class-attribute` `instance-attribute` ¤

OFF = COMPONENT_STATE_OFF

Inverter is switched off.

STANDBY `class-attribute` `instance-attribute` ¤

STANDBY = COMPONENT_STATE_STANDBY

The PbInverteris in a standby state, and is disconnected from the grid.

When connected to the grid, it run a few tests, and move to the IDLE state.

SWITCHING_OFF `class-attribute` `instance-attribute` ¤

SWITCHING_OFF = COMPONENT_STATE_SWITCHING_OFF

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

SWITCHING_ON `class-attribute` `instance-attribute` ¤

SWITCHING_ON = COMPONENT_STATE_SWITCHING_ON

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

UNAVAILABLE `class-attribute` `instance-attribute` ¤

UNAVAILABLE = COMPONENT_STATE_UNAVAILABLE

The inverter is online, but currently unavailable.

Possibly due to a pre- scheduled maintenance.

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: inverter_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.InverterData `dataclass` ¤

Bases: ComponentData

A wrapper class for holding inverter data.

Source code in frequenz/client/microgrid/_component_data.py

@dataclass(frozen=True)
class InverterData(ComponentData):  # pylint: disable=too-many-instance-attributes
    """A wrapper class for holding inverter 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.
    """

    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).
    """

    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.
    """

    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 inverter power in watts.

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

    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 inverter power in watts.

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

    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 inverter power in watts.

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

    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 inverter power in watts.

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

    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)."""

    component_state: InverterComponentState
    """State of the inverter."""

    errors: list[InverterError]
    """List of errors from the component."""

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

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

        Returns:
            Instance of InverterData created from the protobuf message.
        """
        raw_power = raw.inverter.data.ac.power_active
        inverter_data = cls(
            component_id=raw.id,
            timestamp=raw.ts.ToDatetime(tzinfo=timezone.utc),
            active_power=raw.inverter.data.ac.power_active.value,
            active_power_per_phase=(
                raw.inverter.data.ac.phase_1.power_active.value,
                raw.inverter.data.ac.phase_2.power_active.value,
                raw.inverter.data.ac.phase_3.power_active.value,
            ),
            reactive_power=raw.inverter.data.ac.power_reactive.value,
            reactive_power_per_phase=(
                raw.inverter.data.ac.phase_1.power_reactive.value,
                raw.inverter.data.ac.phase_2.power_reactive.value,
                raw.inverter.data.ac.phase_3.power_reactive.value,
            ),
            current_per_phase=(
                raw.inverter.data.ac.phase_1.current.value,
                raw.inverter.data.ac.phase_2.current.value,
                raw.inverter.data.ac.phase_3.current.value,
            ),
            voltage_per_phase=(
                raw.inverter.data.ac.phase_1.voltage.value,
                raw.inverter.data.ac.phase_2.voltage.value,
                raw.inverter.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,
            frequency=raw.inverter.data.ac.frequency.value,
            component_state=InverterComponentState.from_pb(
                raw.inverter.state.component_state
            ),
            errors=[InverterError.from_pb(e) for e in raw.inverter.errors],
        )

        inverter_data._set_raw(raw=raw)
        return inverter_data

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 inverter power in watts.

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

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 inverter power in watts.

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

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 inverter power in watts.

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

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 inverter power in watts.

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

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.

component_id `instance-attribute` ¤

component_id: int

The ID identifying this component in the microgrid.

component_state `instance-attribute` ¤

component_state: InverterComponentState

State of the inverter.

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.

errors `instance-attribute` ¤

errors: list[InverterError]

List of errors from the component.

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: 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 `instance-attribute` ¤

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).

timestamp `instance-attribute` ¤

timestamp: datetime

The timestamp of when the data was measured.

voltage_per_phase `instance-attribute` ¤

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.

Functions¤

from_proto `classmethod` ¤

from_proto(raw: ComponentData) -> Self

Create InverterData from a protobuf message.

PARAMETER	DESCRIPTION
`raw`	raw component data as decoded from the wire. TYPE: `ComponentData`

RETURNS	DESCRIPTION
`Self`	Instance of InverterData 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 InverterData from a protobuf message.

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

    Returns:
        Instance of InverterData created from the protobuf message.
    """
    raw_power = raw.inverter.data.ac.power_active
    inverter_data = cls(
        component_id=raw.id,
        timestamp=raw.ts.ToDatetime(tzinfo=timezone.utc),
        active_power=raw.inverter.data.ac.power_active.value,
        active_power_per_phase=(
            raw.inverter.data.ac.phase_1.power_active.value,
            raw.inverter.data.ac.phase_2.power_active.value,
            raw.inverter.data.ac.phase_3.power_active.value,
        ),
        reactive_power=raw.inverter.data.ac.power_reactive.value,
        reactive_power_per_phase=(
            raw.inverter.data.ac.phase_1.power_reactive.value,
            raw.inverter.data.ac.phase_2.power_reactive.value,
            raw.inverter.data.ac.phase_3.power_reactive.value,
        ),
        current_per_phase=(
            raw.inverter.data.ac.phase_1.current.value,
            raw.inverter.data.ac.phase_2.current.value,
            raw.inverter.data.ac.phase_3.current.value,
        ),
        voltage_per_phase=(
            raw.inverter.data.ac.phase_1.voltage.value,
            raw.inverter.data.ac.phase_2.voltage.value,
            raw.inverter.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,
        frequency=raw.inverter.data.ac.frequency.value,
        component_state=InverterComponentState.from_pb(
            raw.inverter.state.component_state
        ),
        errors=[InverterError.from_pb(e) for e in raw.inverter.errors],
    )

    inverter_data._set_raw(raw=raw)
    return inverter_data

frequenz.client.microgrid.InverterError `dataclass` ¤

An inverter error.

Source code in frequenz/client/microgrid/_component_error.py

@dataclass(frozen=True, kw_only=True)
class InverterError:
    """An inverter error."""

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

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

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

    @classmethod
    def from_pb(cls, raw: inverter_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=InverterErrorCode.from_pb(raw.code),
            level=ErrorLevel.from_pb(raw.level),
            message=raw.msg,
        )

Attributes¤

code `class-attribute` `instance-attribute` ¤

code: InverterErrorCode = UNSPECIFIED

The error code.

level `class-attribute` `instance-attribute` ¤

level: ErrorLevel = UNSPECIFIED

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: inverter_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=InverterErrorCode.from_pb(raw.code),
        level=ErrorLevel.from_pb(raw.level),
        message=raw.msg,
    )

frequenz.client.microgrid.InverterErrorCode ¤

Bases: Enum

Inverter error code.

Source code in frequenz/client/microgrid/_component_error.py

class InverterErrorCode(Enum):
    """Inverter error code."""

    UNSPECIFIED = inverter_pb2.ErrorCode.ERROR_CODE_UNSPECIFIED
    """Unspecified inverter error code."""

    @classmethod
    def from_pb(cls, code: inverter_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¤

UNSPECIFIED `class-attribute` `instance-attribute` ¤

UNSPECIFIED = ERROR_CODE_UNSPECIFIED

Unspecified inverter 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: inverter_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.InverterType ¤

Bases: ComponentType

Enum representing inverter types.

Source code in frequenz/client/microgrid/_component.py

class InverterType(ComponentType):
    """Enum representing inverter types."""

    NONE = inverter_pb2.Type.TYPE_UNSPECIFIED
    """Unspecified inverter type."""

    BATTERY = inverter_pb2.Type.TYPE_BATTERY
    """Battery inverter."""

    SOLAR = inverter_pb2.Type.TYPE_SOLAR
    """Solar inverter."""

    HYBRID = inverter_pb2.Type.TYPE_HYBRID
    """Hybrid inverter."""

Attributes¤

BATTERY `class-attribute` `instance-attribute` ¤

BATTERY = TYPE_BATTERY

Battery inverter.

HYBRID `class-attribute` `instance-attribute` ¤

HYBRID = TYPE_HYBRID

Hybrid inverter.

NONE `class-attribute` `instance-attribute` ¤

NONE = TYPE_UNSPECIFIED

Unspecified inverter type.

SOLAR `class-attribute` `instance-attribute` ¤

SOLAR = TYPE_SOLAR

Solar inverter.

frequenz.client.microgrid.Location `dataclass` ¤

Metadata for the location of microgrid.

Source code in frequenz/client/microgrid/_metadata.py

@dataclass(frozen=True, kw_only=True)
class Location:
    """Metadata for the location of microgrid."""

    latitude: float | None = None
    """The latitude of the microgrid in degree."""

    longitude: float | None = None
    """The longitude of the microgrid in degree."""

    timezone: ZoneInfo | None = None
    """The timezone of the microgrid.

    If not passed during construction (or `None` is passed), and there is a `longitude`
    and `latitude`, then the timezone wil be looked up in a database based on the
    coordinates. This lookup could fail, in which case the timezone will still be
    `None`.
    """

    def __post_init__(self) -> None:
        """Initialize the timezone of the microgrid."""
        if self.latitude is None or self.longitude is None or self.timezone is not None:
            return

        timezone = _timezone_finder.timezone_at(lat=self.latitude, lng=self.longitude)
        if timezone:
            # The dataclass is frozen, so it needs to use __setattr__ to set the timezone.
            object.__setattr__(self, "timezone", ZoneInfo(key=timezone))

Attributes¤

latitude `class-attribute` `instance-attribute` ¤

latitude: float | None = None

The latitude of the microgrid in degree.

longitude `class-attribute` `instance-attribute` ¤

longitude: float | None = None

The longitude of the microgrid in degree.

timezone `class-attribute` `instance-attribute` ¤

timezone: ZoneInfo | None = None

The timezone of the microgrid.

If not passed during construction (or None is passed), and there is a longitude and latitude, then the timezone wil be looked up in a database based on the coordinates. This lookup could fail, in which case the timezone will still be None.

Functions¤

__post_init__ ¤

__post_init__() -> None

Initialize the timezone of the microgrid.

Source code in frequenz/client/microgrid/_metadata.py

def __post_init__(self) -> None:
    """Initialize the timezone of the microgrid."""
    if self.latitude is None or self.longitude is None or self.timezone is not None:
        return

    timezone = _timezone_finder.timezone_at(lat=self.latitude, lng=self.longitude)
    if timezone:
        # The dataclass is frozen, so it needs to use __setattr__ to set the timezone.
        object.__setattr__(self, "timezone", ZoneInfo(key=timezone))

frequenz.client.microgrid.Metadata `dataclass` ¤

Metadata for the microgrid.

Source code in frequenz/client/microgrid/_metadata.py

@dataclass(frozen=True, kw_only=True)
class Metadata:
    """Metadata for the microgrid."""

    microgrid_id: int | None = None
    """The ID of the microgrid."""

    location: Location | None = None
    """The location of the microgrid."""

Attributes¤

location `class-attribute` `instance-attribute` ¤

location: Location | None = None

The location of the microgrid.

microgrid_id `class-attribute` `instance-attribute` ¤

microgrid_id: int | None = None

The ID of the microgrid.

frequenz.client.microgrid.MeterData `dataclass` ¤

Bases: ComponentData

A wrapper class for holding meter data.

Source code in frequenz/client/microgrid/_component_data.py

@dataclass(frozen=True)
class MeterData(ComponentData):
    """A wrapper class for holding meter 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.
    """

    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).
    """

    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.
    """

    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.
    """

    frequency: float
    """The AC power frequency in Hertz (Hz)."""

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

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

        Returns:
            Instance of MeterData created from the protobuf message.
        """
        meter_data = cls(
            component_id=raw.id,
            timestamp=raw.ts.ToDatetime(tzinfo=timezone.utc),
            active_power=raw.meter.data.ac.power_active.value,
            active_power_per_phase=(
                raw.meter.data.ac.phase_1.power_active.value,
                raw.meter.data.ac.phase_2.power_active.value,
                raw.meter.data.ac.phase_3.power_active.value,
            ),
            reactive_power=raw.meter.data.ac.power_reactive.value,
            reactive_power_per_phase=(
                raw.meter.data.ac.phase_1.power_reactive.value,
                raw.meter.data.ac.phase_2.power_reactive.value,
                raw.meter.data.ac.phase_3.power_reactive.value,
            ),
            current_per_phase=(
                raw.meter.data.ac.phase_1.current.value,
                raw.meter.data.ac.phase_2.current.value,
                raw.meter.data.ac.phase_3.current.value,
            ),
            voltage_per_phase=(
                raw.meter.data.ac.phase_1.voltage.value,
                raw.meter.data.ac.phase_2.voltage.value,
                raw.meter.data.ac.phase_3.voltage.value,
            ),
            frequency=raw.meter.data.ac.frequency.value,
        )
        meter_data._set_raw(raw=raw)
        return meter_data

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_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.

component_id `instance-attribute` ¤

component_id: int

The ID identifying this component in the microgrid.

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

The AC power 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: 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 `instance-attribute` ¤

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).

timestamp `instance-attribute` ¤

timestamp: datetime

The timestamp of when the data was measured.

voltage_per_phase `instance-attribute` ¤

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.

Functions¤

from_proto `classmethod` ¤

from_proto(raw: ComponentData) -> Self

Create MeterData from a protobuf message.

PARAMETER	DESCRIPTION
`raw`	raw component data as decoded from the wire. TYPE: `ComponentData`

RETURNS	DESCRIPTION
`Self`	Instance of MeterData 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 MeterData from a protobuf message.

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

    Returns:
        Instance of MeterData created from the protobuf message.
    """
    meter_data = cls(
        component_id=raw.id,
        timestamp=raw.ts.ToDatetime(tzinfo=timezone.utc),
        active_power=raw.meter.data.ac.power_active.value,
        active_power_per_phase=(
            raw.meter.data.ac.phase_1.power_active.value,
            raw.meter.data.ac.phase_2.power_active.value,
            raw.meter.data.ac.phase_3.power_active.value,
        ),
        reactive_power=raw.meter.data.ac.power_reactive.value,
        reactive_power_per_phase=(
            raw.meter.data.ac.phase_1.power_reactive.value,
            raw.meter.data.ac.phase_2.power_reactive.value,
            raw.meter.data.ac.phase_3.power_reactive.value,
        ),
        current_per_phase=(
            raw.meter.data.ac.phase_1.current.value,
            raw.meter.data.ac.phase_2.current.value,
            raw.meter.data.ac.phase_3.current.value,
        ),
        voltage_per_phase=(
            raw.meter.data.ac.phase_1.voltage.value,
            raw.meter.data.ac.phase_2.voltage.value,
            raw.meter.data.ac.phase_3.voltage.value,
        ),
        frequency=raw.meter.data.ac.frequency.value,
    )
    meter_data._set_raw(raw=raw)
    return meter_data

frequenz.client.microgrid.OperationAborted ¤

Bases: GrpcError

The operation was aborted.

Typically due to a concurrency issue or transaction abort.

Source code in frequenz/client/base/exception.py

class OperationAborted(GrpcError):
    """The operation was aborted.

    Typically due to a concurrency issue or transaction abort.
    """

    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="The operation was aborted",
            grpc_error=grpc_error,
            retryable=True,
        )

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="The operation was aborted",
        grpc_error=grpc_error,
        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.OperationCancelled ¤

Bases: GrpcError

The operation was cancelled.

Source code in frequenz/client/base/exception.py

class OperationCancelled(GrpcError):
    """The operation was cancelled."""

    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="The operation was cancelled",
            grpc_error=grpc_error,
            retryable=True,
        )

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="The operation was cancelled",
        grpc_error=grpc_error,
        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.OperationNotImplemented ¤

Bases: GrpcError

The operation is not implemented or not supported/enabled in this service.

Source code in frequenz/client/base/exception.py

class OperationNotImplemented(GrpcError):
    """The operation is not implemented or not supported/enabled in this service."""

    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="The operation is not implemented or not supported/enabled in "
            "this service",
            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="The operation is not implemented or not supported/enabled in "
        "this service",
        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.OperationOutOfRange ¤

Bases: GrpcError

The operation was attempted past the valid range.

Unlike InvalidArgument, this error indicates a problem that may be fixed if the system state changes.

There is a fair bit of overlap with OperationPreconditionFailed, this error is just a more specific version of that error and could be the result of an operation that doesn't even take any arguments.

Source code in frequenz/client/base/exception.py

class OperationOutOfRange(GrpcError):
    """The operation was attempted past the valid range.

    Unlike [InvalidArgument][frequenz.client.base.exception.InvalidArgument], this error
    indicates a problem that may be fixed if the system state changes.

    There is a fair bit of overlap with
    [OperationPreconditionFailed][frequenz.client.base.exception.OperationPreconditionFailed],
    this error is just a more specific version of that error and could be the result of
    an operation that doesn't even take any arguments.
    """

    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="The operation was attempted past the valid range",
            grpc_error=grpc_error,
            retryable=True,  # If the system state changes it might succeed
        )

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="The operation was attempted past the valid range",
        grpc_error=grpc_error,
        retryable=True,  # If the system state changes it 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.OperationPreconditionFailed ¤

Bases: GrpcError

The operation was rejected because the system is not in a required state.

For example, the directory to be deleted is non-empty, an rmdir operation is applied to a non-directory, etc. The user should perform some corrective action before retrying the operation.

Source code in frequenz/client/base/exception.py

class OperationPreconditionFailed(GrpcError):
    """The operation was rejected because the system is not in a required state.

    For example, the directory to be deleted is non-empty, an rmdir operation is applied
    to a non-directory, etc. The user should perform some corrective action before
    retrying the operation.
    """

    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="The operation was rejected because the system is not in a "
            "required state",
            grpc_error=grpc_error,
            retryable=True,  # If the system state changes it might succeed
        )

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="The operation was rejected because the system is not in a "
        "required state",
        grpc_error=grpc_error,
        retryable=True,  # If the system state changes it 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.OperationTimedOut ¤

Bases: GrpcError

The time limit was exceeded while waiting for the operationt o complete.

For operations that change the state of the system, this error may be returned even if the operation has completed successfully. For example, a successful response from a server could have been delayed long.

Source code in frequenz/client/base/exception.py

class OperationTimedOut(GrpcError):
    """The time limit was exceeded while waiting for the operationt o complete.

    For operations that change the state of the system, this error may be returned even
    if the operation has completed successfully. For example, a successful response from
    a server could have been delayed long.
    """

    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="The time limit was exceeded while waiting for the operation "
            "to complete",
            grpc_error=grpc_error,
            retryable=True,
        )

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="The time limit was exceeded while waiting for the operation "
        "to complete",
        grpc_error=grpc_error,
        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.OperationUnauthenticated ¤

Bases: GrpcError

The request does not have valid authentication credentials for the operation.

Source code in frequenz/client/base/exception.py

class OperationUnauthenticated(GrpcError):
    """The request does not have valid authentication credentials for the operation."""

    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="The request does not have valid authentication credentials "
            "for the operation",
            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="The request does not have valid authentication credentials "
        "for the operation",
        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.PermissionDenied ¤

Bases: GrpcError

The caller does not have permission to execute the specified operation.

Note that when the operation is rejected due to other reasons, such as the resources being exhausted or the user not being authenticated at all, different errors should be catched instead (ResourceExhausted and OperationUnauthenticated respectively).

Source code in frequenz/client/base/exception.py

class PermissionDenied(GrpcError):
    """The caller does not have permission to execute the specified operation.

    Note that when the operation is rejected due to other reasons, such as the resources
    being exhausted or the user not being authenticated at all, different errors should
    be catched instead
    ([ResourceExhausted][frequenz.client.base.exception.ResourceExhausted] and
    [OperationUnauthenticated][frequenz.client.base.exception.OperationUnauthenticated]
    respectively).
    """

    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="The caller does not have permission to execute the specified "
            "operation",
            grpc_error=grpc_error,
            retryable=True,  # If the user is granted permission it might succeed
        )

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="The caller does not have permission to execute the specified "
        "operation",
        grpc_error=grpc_error,
        retryable=True,  # If the user is granted permission it 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.ResourceExhausted ¤

Bases: GrpcError

Some resource has been exhausted (for example per-user quota, disk space, etc.).

Source code in frequenz/client/base/exception.py

class ResourceExhausted(GrpcError):
    """Some resource has been exhausted (for example per-user quota, disk space, etc.)."""

    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="Some resource has been exhausted (for example per-user quota, "
            "disk space, etc.)",
            grpc_error=grpc_error,
            retryable=True,  # If the resource is freed it might succeed
        )

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="Some resource has been exhausted (for example per-user quota, "
        "disk space, etc.)",
        grpc_error=grpc_error,
        retryable=True,  # If the resource is freed it 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.ServiceUnavailable ¤

Bases: GrpcError

The service is currently unavailable.

This is most likely a transient condition, which can be corrected by retrying with a backoff. Note that it is not always safe to retry non-idempotent operations.

Source code in frequenz/client/base/exception.py

class ServiceUnavailable(GrpcError):
    """The service is currently unavailable.

    This is most likely a transient condition, which can be corrected by retrying with
    a backoff. Note that it is not always safe to retry non-idempotent operations.
    """

    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="The service is currently unavailable",
            grpc_error=grpc_error,
            retryable=True,  # If the service becomes available it might succeed
        )

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="The service is currently unavailable",
        grpc_error=grpc_error,
        retryable=True,  # If the service becomes available it 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.UnknownError ¤

Bases: GrpcError

There was an error that can't be described using other statuses.

Source code in frequenz/client/base/exception.py

class UnknownError(GrpcError):
    """There was an error that can't be described using other statuses."""

    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="There was an error that can't be described using other statuses",
            grpc_error=grpc_error,
            retryable=True,  # We don't know so we assume it's retryable
        )

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="There was an error that can't be described using other statuses",
        grpc_error=grpc_error,
        retryable=True,  # We don't know so we assume it's retryable
    )

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.UnrecognizedGrpcStatus ¤

Bases: GrpcError

The gRPC server returned an unrecognized status code.

Source code in frequenz/client/base/exception.py

class UnrecognizedGrpcStatus(GrpcError):
    """The gRPC server returned an unrecognized status code."""

    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="Got an unrecognized status code",
            grpc_error=grpc_error,
            retryable=True,  # We don't know so we assume it's retryable
        )

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="Got an unrecognized status code",
        grpc_error=grpc_error,
        retryable=True,  # We don't know so we assume it's retryable
    )

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,
    )

microgrid

frequenz.client.microgrid ¤

Classes¤

frequenz.client.microgrid.ApiClient ¤

Attributes¤

api instance-attribute ¤

server_url property ¤

Functions¤

__init__ ¤

battery_data async ¤

components async ¤

connections async ¤

ev_charger_data async ¤

inverter_data async ¤

metadata async ¤

meter_data async ¤

set_bounds async ¤

set_power async ¤

set_reactive_power async ¤

frequenz.client.microgrid.ApiClientError ¤

Attributes¤

description instance-attribute ¤

is_retryable instance-attribute ¤

operation instance-attribute ¤

server_url instance-attribute ¤

Functions¤

__init__ ¤

from_grpc_error classmethod ¤

frequenz.client.microgrid.BatteryComponentState ¤

Attributes¤

CHARGING class-attribute instance-attribute ¤

DISCHARGING class-attribute instance-attribute ¤

ERROR class-attribute instance-attribute ¤

IDLE class-attribute instance-attribute ¤

LOCKED class-attribute instance-attribute ¤

OFF class-attribute instance-attribute ¤

SWITCHING_OFF class-attribute instance-attribute ¤

SWITCHING_ON class-attribute instance-attribute ¤

UNKNOWN class-attribute instance-attribute ¤

UNSPECIFIED class-attribute instance-attribute ¤

Functions¤

from_pb classmethod ¤

frequenz.client.microgrid.BatteryData dataclass ¤

Attributes¤

capacity instance-attribute ¤

component_id instance-attribute ¤

component_state instance-attribute ¤

errors instance-attribute ¤

power_exclusion_lower_bound instance-attribute ¤

power_exclusion_upper_bound instance-attribute ¤

power_inclusion_lower_bound instance-attribute ¤

power_inclusion_upper_bound instance-attribute ¤

raw class-attribute instance-attribute ¤

relay_state instance-attribute ¤

soc instance-attribute ¤

soc_lower_bound instance-attribute ¤

soc_upper_bound instance-attribute ¤

temperature instance-attribute ¤

timestamp instance-attribute ¤

Functions¤

from_proto classmethod ¤

frequenz.client.microgrid.BatteryError dataclass ¤

Attributes¤

code class-attribute instance-attribute ¤

level class-attribute instance-attribute ¤

message class-attribute instance-attribute ¤

Functions¤

from_pb classmethod ¤

frequenz.client.microgrid.BatteryErrorCode ¤

Attributes¤

BLOCK_ERROR class-attribute instance-attribute ¤

CALIBRATION_NEEDED class-attribute instance-attribute ¤

CONTROLLER_ERROR class-attribute instance-attribute ¤

EXCEEDED_SOP_CHARGE class-attribute instance-attribute ¤

EXCEEDED_SOP_DISCHARGE class-attribute instance-attribute ¤

EXTERNAL_POWER_SWITCH_ERROR class-attribute instance-attribute ¤

FUSE_ERROR class-attribute instance-attribute ¤

HIGH_CURRENT_CHARGE class-attribute instance-attribute ¤

HIGH_CURRENT_DISCHARGE class-attribute instance-attribute ¤

HIGH_HUMIDITY class-attribute instance-attribute ¤

api `instance-attribute` ¤

server_url `property` ¤

init ¤

battery_data `async` ¤

components `async` ¤

connections `async` ¤

ev_charger_data `async` ¤

inverter_data `async` ¤

metadata `async` ¤

meter_data `async` ¤

set_bounds `async` ¤

set_power `async` ¤

set_reactive_power `async` ¤

description `instance-attribute` ¤

is_retryable `instance-attribute` ¤

operation `instance-attribute` ¤

server_url `instance-attribute` ¤

init ¤

from_grpc_error `classmethod` ¤

CHARGING `class-attribute` `instance-attribute` ¤

DISCHARGING `class-attribute` `instance-attribute` ¤

ERROR `class-attribute` `instance-attribute` ¤

IDLE `class-attribute` `instance-attribute` ¤

LOCKED `class-attribute` `instance-attribute` ¤

OFF `class-attribute` `instance-attribute` ¤

SWITCHING_OFF `class-attribute` `instance-attribute` ¤

SWITCHING_ON `class-attribute` `instance-attribute` ¤

UNKNOWN `class-attribute` `instance-attribute` ¤

UNSPECIFIED `class-attribute` `instance-attribute` ¤

from_pb `classmethod` ¤

frequenz.client.microgrid.BatteryData `dataclass` ¤

capacity `instance-attribute` ¤

component_id `instance-attribute` ¤

component_state `instance-attribute` ¤

errors `instance-attribute` ¤

power_exclusion_lower_bound `instance-attribute` ¤

power_exclusion_upper_bound `instance-attribute` ¤

power_inclusion_lower_bound `instance-attribute` ¤

power_inclusion_upper_bound `instance-attribute` ¤

raw `class-attribute` `instance-attribute` ¤

relay_state `instance-attribute` ¤

soc `instance-attribute` ¤

soc_lower_bound `instance-attribute` ¤

soc_upper_bound `instance-attribute` ¤

temperature `instance-attribute` ¤

timestamp `instance-attribute` ¤

from_proto `classmethod` ¤

frequenz.client.microgrid.BatteryError `dataclass` ¤

code `class-attribute` `instance-attribute` ¤

level `class-attribute` `instance-attribute` ¤

message `class-attribute` `instance-attribute` ¤

from_pb `classmethod` ¤

BLOCK_ERROR `class-attribute` `instance-attribute` ¤

CALIBRATION_NEEDED `class-attribute` `instance-attribute` ¤

CONTROLLER_ERROR `class-attribute` `instance-attribute` ¤

EXCEEDED_SOP_CHARGE `class-attribute` `instance-attribute` ¤

EXCEEDED_SOP_DISCHARGE `class-attribute` `instance-attribute` ¤

EXTERNAL_POWER_SWITCH_ERROR `class-attribute` `instance-attribute` ¤

FUSE_ERROR `class-attribute` `instance-attribute` ¤

HIGH_CURRENT_CHARGE `class-attribute` `instance-attribute` ¤

HIGH_CURRENT_DISCHARGE `class-attribute` `instance-attribute` ¤

HIGH_HUMIDITY `class-attribute` `instance-attribute` ¤

HIGH_TEMPERATURE `class-attribute` `instance-attribute` ¤

HIGH_VOLTAGE `class-attribute` `instance-attribute` ¤

LOW_SOH `class-attribute` `instance-attribute` ¤

LOW_TEMPERATURE `class-attribute` `instance-attribute` ¤

LOW_VOLTAGE `class-attribute` `instance-attribute` ¤

PRECHARGE_ERROR `class-attribute` `instance-attribute` ¤

RELAY_CYCLE_LIMIT_REACHED `class-attribute` `instance-attribute` ¤

RELAY_ERROR `class-attribute` `instance-attribute` ¤

SYSTEM_IMBALANCE `class-attribute` `instance-attribute` ¤

SYSTEM_PLAUSIBILITY_ERROR `class-attribute` `instance-attribute` ¤

SYSTEM_UNDERVOLTAGE_SHUTDOWN `class-attribute` `instance-attribute` ¤

UNSPECIFIED `class-attribute` `instance-attribute` ¤

from_pb `classmethod` ¤

CLOSED `class-attribute` `instance-attribute` ¤

ERROR `class-attribute` `instance-attribute` ¤

LOCKED `class-attribute` `instance-attribute` ¤

OPENED `class-attribute` `instance-attribute` ¤

PRECHARGING `class-attribute` `instance-attribute` ¤