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_bounds(
        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(
    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

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