Skip to content

Index

frequenz.client.microgrid ¤

Client to connect to the Microgrid API.

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

Attributes¤

frequenz.client.microgrid.DEFAULT_CHANNEL_OPTIONS module-attribute ¤

DEFAULT_CHANNEL_OPTIONS = replace(
    ChannelOptions(), ssl=SslOptions(enabled=False)
)

The default channel options for the microgrid API client.

These are the same defaults as the common default options but with SSL disabled, as the microgrid API does not use SSL by default.

frequenz.client.microgrid.DEFAULT_GRPC_CALL_TIMEOUT module-attribute ¤

DEFAULT_GRPC_CALL_TIMEOUT = 60.0

The default timeout for gRPC calls made by this client (in seconds).

Classes¤

frequenz.client.microgrid.ApiClientError ¤

Bases: Exception

There was an error in an API client.

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

The following sub-classes are available:

Source code in frequenz/client/base/exception.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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.ClientNotConnected ¤

Bases: ApiClientError

The client is not connected to the server.

Source code in frequenz/client/base/exception.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
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 
120
121
122
123
124
125
126
127
128
129
130
131
132
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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.DataLoss ¤

Bases: GrpcError

Unrecoverable data loss or corruption.

Source code in frequenz/client/base/exception.py 
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
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 
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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.DeliveryArea dataclass ¤

A geographical or administrative region where electricity deliveries occur.

DeliveryArea represents the geographical or administrative region, usually defined and maintained by a Transmission System Operator (TSO), where electricity deliveries for a contract occur.

The concept is important to energy trading as it delineates the agreed-upon delivery location. Delivery areas can have different codes based on the jurisdiction in which they operate.

Jurisdictional Differences

This is typically represented by specific codes according to local jurisdiction.

In Europe, this is represented by an EIC (Energy Identification Code). List of EICs.

Source code in frequenz/client/microgrid/_delivery_area.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@dataclass(frozen=True, kw_only=True)
class DeliveryArea:
    """A geographical or administrative region where electricity deliveries occur.

    DeliveryArea represents the geographical or administrative region, usually defined
    and maintained by a Transmission System Operator (TSO), where electricity deliveries
    for a contract occur.

    The concept is important to energy trading as it delineates the agreed-upon delivery
    location. Delivery areas can have different codes based on the jurisdiction in
    which they operate.

    Note: Jurisdictional Differences
        This is typically represented by specific codes according to local jurisdiction.

        In Europe, this is represented by an
        [EIC](https://en.wikipedia.org/wiki/Energy_Identification_Code) (Energy
        Identification Code). [List of
        EICs](https://www.entsoe.eu/data/energy-identification-codes-eic/eic-approved-codes/).
    """

    code: str | None
    """The code representing the unique identifier for the delivery area."""

    code_type: EnergyMarketCodeType | int
    """Type of code used for identifying the delivery area itself.

    This code could be extended in the future, in case an unknown code type is
    encountered, a plain integer value is used to represent it.
    """

    def __str__(self) -> str:
        """Return a human-readable string representation of this instance."""
        code = self.code or "<NO CODE>"
        code_type = (
            f"type={self.code_type}"
            if isinstance(self.code_type, int)
            else self.code_type.name
        )
        return f"{code}[{code_type}]"
Attributes¤
code instance-attribute ¤
code: str | None

The code representing the unique identifier for the delivery area.

code_type instance-attribute ¤
code_type: EnergyMarketCodeType | int

Type of code used for identifying the delivery area itself.

This code could be extended in the future, in case an unknown code type is encountered, a plain integer value is used to represent it.

Functions¤
__str__ ¤
__str__() -> str

Return a human-readable string representation of this instance.

Source code in frequenz/client/microgrid/_delivery_area.py 
81
82
83
84
85
86
87
88
89
def __str__(self) -> str:
    """Return a human-readable string representation of this instance."""
    code = self.code or "<NO CODE>"
    code_type = (
        f"type={self.code_type}"
        if isinstance(self.code_type, int)
        else self.code_type.name
    )
    return f"{code}[{code_type}]"

frequenz.client.microgrid.EnergyMarketCodeType ¤

Bases: Enum

The identification code types used in the energy market.

CodeType specifies the type of identification code used for uniquely identifying various entities such as delivery areas, market participants, and grid components within the energy market.

This enumeration aims to offer compatibility across different jurisdictional standards.

Understanding Code Types

Different regions or countries may have their own standards for uniquely identifying various entities within the energy market. For example, in Europe, the Energy Identification Code (EIC) is commonly used for this purpose.

Extensibility

New code types can be added to this enum to accommodate additional regional standards, enhancing the API's adaptability.

Validation Required

The chosen code type should correspond correctly with the code field in the relevant message objects, such as DeliveryArea or Counterparty. Failure to match the code type with the correct code could lead to processing errors.

Source code in frequenz/client/microgrid/_delivery_area.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@enum.unique
class EnergyMarketCodeType(enum.Enum):
    """The identification code types used in the energy market.

    CodeType specifies the type of identification code used for uniquely
    identifying various entities such as delivery areas, market participants,
    and grid components within the energy market.

    This enumeration aims to
    offer compatibility across different jurisdictional standards.

    Note: Understanding Code Types
        Different regions or countries may have their own standards for uniquely
        identifying various entities within the energy market. For example, in
        Europe, the Energy Identification Code (EIC) is commonly used for this
        purpose.

    Note: Extensibility
        New code types can be added to this enum to accommodate additional regional
        standards, enhancing the API's adaptability.

    Danger: Validation Required
        The chosen code type should correspond correctly with the `code` field in
        the relevant message objects, such as `DeliveryArea` or `Counterparty`.
        Failure to match the code type with the correct code could lead to
        processing errors.
    """

    UNSPECIFIED = delivery_area_pb2.ENERGY_MARKET_CODE_TYPE_UNSPECIFIED
    """Unspecified type. This value is a placeholder and should not be used."""

    EUROPE_EIC = delivery_area_pb2.ENERGY_MARKET_CODE_TYPE_EUROPE_EIC
    """European Energy Identification Code Standard."""

    US_NERC = delivery_area_pb2.ENERGY_MARKET_CODE_TYPE_US_NERC
    """North American Electric Reliability Corporation identifiers."""
Attributes¤
EUROPE_EIC class-attribute instance-attribute ¤
EUROPE_EIC = ENERGY_MARKET_CODE_TYPE_EUROPE_EIC

European Energy Identification Code Standard.

UNSPECIFIED class-attribute instance-attribute ¤
UNSPECIFIED = ENERGY_MARKET_CODE_TYPE_UNSPECIFIED

Unspecified type. This value is a placeholder and should not be used.

US_NERC class-attribute instance-attribute ¤
US_NERC = ENERGY_MARKET_CODE_TYPE_US_NERC

North American Electric Reliability Corporation identifiers.

frequenz.client.microgrid.EntityAlreadyExists ¤

Bases: GrpcError

The entity that we attempted to create already exists.

Source code in frequenz/client/base/exception.py 
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
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 
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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 
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
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 
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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.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:

References
Source code in frequenz/client/base/exception.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
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 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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 
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
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 
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
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 
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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.Lifetime dataclass ¤

An active operational period of a microgrid asset.

Warning

The end timestamp indicates that the asset has been permanently removed from the system.

Source code in frequenz/client/microgrid/_lifetime.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@dataclass(frozen=True, kw_only=True)
class Lifetime:
    """An active operational period of a microgrid asset.

    Warning:
        The [`end`][frequenz.client.microgrid.Lifetime.end] timestamp indicates that the
        asset has been permanently removed from the system.
    """

    start: datetime | None = None
    """The moment when the asset became operationally active.

    If `None`, the asset is considered to be active in any past moment previous to the
    [`end`][frequenz.client.microgrid.Lifetime.end].
    """

    end: datetime | None = None
    """The moment when the asset's operational activity ceased.

    If `None`, the asset is considered to be active with no plans to be deactivated.
    """

    def __post_init__(self) -> None:
        """Validate this lifetime."""
        if self.start is not None and self.end is not None and self.start > self.end:
            raise ValueError(
                f"Start ({self.start}) must be before or equal to end ({self.end})"
            )

    def is_operational_at(self, timestamp: datetime) -> bool:
        """Check whether this lifetime is active at a specific timestamp."""
        # Handle start time - it's not active if start is in the future
        if self.start is not None and self.start > timestamp:
            return False
        # Handle end time - active up to and including end time
        if self.end is not None:
            return self.end >= timestamp
        # self.end is None, and either self.start is None or self.start <= timestamp,
        # so it is active at this timestamp
        return True

    def is_operational_now(self) -> bool:
        """Whether this lifetime is currently active."""
        return self.is_operational_at(datetime.now(timezone.utc))
Attributes¤
end class-attribute instance-attribute ¤
end: datetime | None = None

The moment when the asset's operational activity ceased.

If None, the asset is considered to be active with no plans to be deactivated.

start class-attribute instance-attribute ¤
start: datetime | None = None

The moment when the asset became operationally active.

If None, the asset is considered to be active in any past moment previous to the end.

Functions¤
__post_init__ ¤
__post_init__() -> None

Validate this lifetime.

Source code in frequenz/client/microgrid/_lifetime.py 
33
34
35
36
37
38
def __post_init__(self) -> None:
    """Validate this lifetime."""
    if self.start is not None and self.end is not None and self.start > self.end:
        raise ValueError(
            f"Start ({self.start}) must be before or equal to end ({self.end})"
        )
is_operational_at ¤
is_operational_at(timestamp: datetime) -> bool

Check whether this lifetime is active at a specific timestamp.

Source code in frequenz/client/microgrid/_lifetime.py 
40
41
42
43
44
45
46
47
48
49
50
def is_operational_at(self, timestamp: datetime) -> bool:
    """Check whether this lifetime is active at a specific timestamp."""
    # Handle start time - it's not active if start is in the future
    if self.start is not None and self.start > timestamp:
        return False
    # Handle end time - active up to and including end time
    if self.end is not None:
        return self.end >= timestamp
    # self.end is None, and either self.start is None or self.start <= timestamp,
    # so it is active at this timestamp
    return True
is_operational_now ¤
is_operational_now() -> bool

Whether this lifetime is currently active.

Source code in frequenz/client/microgrid/_lifetime.py 
52
53
54
def is_operational_now(self) -> bool:
    """Whether this lifetime is currently active."""
    return self.is_operational_at(datetime.now(timezone.utc))

frequenz.client.microgrid.Location dataclass ¤

A location of a microgrid.

Source code in frequenz/client/microgrid/_location.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
@dataclass(frozen=True, kw_only=True)
class Location:
    """A location of a microgrid."""

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

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

    country_code: str | None
    """The country code of the microgrid in ISO 3166-1 Alpha 2 format."""

    def __str__(self) -> str:
        """Return the short string representation of this instance."""
        country = self.country_code or "<NO COUNTRY CODE>"
        lat = f"{self.latitude:.2f}" if self.latitude is not None else "?"
        lon = f"{self.longitude:.2f}" if self.longitude is not None else "?"
        coordinates = ""
        if self.latitude is not None or self.longitude is not None:
            coordinates = f":({lat}, {lon})"
        return f"{country}{coordinates}"
Attributes¤
country_code instance-attribute ¤
country_code: str | None

The country code of the microgrid in ISO 3166-1 Alpha 2 format.

latitude instance-attribute ¤
latitude: float | None

The latitude of the microgrid in degree.

longitude instance-attribute ¤
longitude: float | None

The longitude of the microgrid in degree.

Functions¤
__str__ ¤
__str__() -> str

Return the short string representation of this instance.

Source code in frequenz/client/microgrid/_location.py 
23
24
25
26
27
28
29
30
31
def __str__(self) -> str:
    """Return the short string representation of this instance."""
    country = self.country_code or "<NO COUNTRY CODE>"
    lat = f"{self.latitude:.2f}" if self.latitude is not None else "?"
    lon = f"{self.longitude:.2f}" if self.longitude is not None else "?"
    coordinates = ""
    if self.latitude is not None or self.longitude is not None:
        coordinates = f":({lat}, {lon})"
    return f"{country}{coordinates}"

frequenz.client.microgrid.MicrogridApiClient ¤

Bases: BaseApiClient[MicrogridStub]

A microgrid API client.

Source code in frequenz/client/microgrid/_client.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
class MicrogridApiClient(client.BaseApiClient[microgrid_pb2_grpc.MicrogridStub]):
    """A microgrid API client."""

    def __init__(
        self,
        server_url: str,
        *,
        channel_defaults: channel.ChannelOptions = DEFAULT_CHANNEL_OPTIONS,
        connect: bool = True,
        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_defaults: The default options use to create the channel when not
                specified in the URL.
            connect: Whether to connect to the server as soon as a client instance is
                created. If `False`, the client will not connect to the server until
                [connect()][frequenz.client.base.client.BaseApiClient.connect] is
                called.
            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.
        """
        super().__init__(
            server_url,
            microgrid_pb2_grpc.MicrogridStub,
            connect=connect,
            channel_defaults=channel_defaults,
        )
        self._component_data_broadcasters: dict[
            str,
            streaming.GrpcStreamBroadcaster[
                microgrid_pb2.ReceiveElectricalComponentTelemetryStreamResponse,
                ComponentDataSamples,
            ],
        ] = {}
        self._sensor_data_broadcasters: dict[
            str,
            streaming.GrpcStreamBroadcaster[Any, Any],
        ] = {}
        self._retry_strategy = retry_strategy

    @property
    def stub(self) -> microgrid_pb2_grpc.MicrogridAsyncStub:
        """The gRPC stub for the API."""
        if self.channel is None or self._stub is None:
            raise ClientNotConnected(server_url=self.server_url, operation="stub")
        # This type: ignore is needed because we need to cast the sync stub to
        # the async stub, but we can't use cast because the async stub doesn't
        # actually exists to the eyes of the interpreter, it only exists for the
        # type-checker, so it can only be used for type hints.
        return self._stub  # type: ignore

    @override
    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: Any | None,
    ) -> bool | None:
        """Close the gRPC channel and stop all broadcasters."""
        exceptions = list(
            exc
            for exc in await asyncio.gather(
                *(
                    broadcaster.stop()
                    for broadcaster in itertools.chain(
                        self._component_data_broadcasters.values(),
                        self._sensor_data_broadcasters.values(),
                    )
                ),
                return_exceptions=True,
            )
            if isinstance(exc, BaseException)
        )
        self._component_data_broadcasters.clear()
        self._sensor_data_broadcasters.clear()

        result = None
        try:
            result = await super().__aexit__(exc_type, exc_val, exc_tb)
        except Exception as exc:  # pylint: disable=broad-except
            exceptions.append(exc)
        if exceptions:
            raise BaseExceptionGroup(
                "Error while disconnecting from the microgrid API", exceptions
            )
        return result

    async def get_microgrid_info(  # noqa: DOC502 (raises ApiClientError indirectly)
        self,
    ) -> MicrogridInfo:
        """Retrieve information about the local microgrid.

        This consists of information about the overall microgrid, for example, the
        microgrid ID and its location.  It does not include information about the
        electrical components or sensors in the microgrid.

        Returns:
            The information about the local microgrid.

        Raises:
            ApiClientError: If there are any errors communicating with the Microgrid API,
                most likely a subclass of
                [GrpcError][frequenz.client.microgrid.GrpcError].
        """
        response = await client.call_stub_method(
            self,
            lambda: self.stub.GetMicrogrid(Empty(), timeout=DEFAULT_GRPC_CALL_TIMEOUT),
            method_name="GetMicrogridMetadata",
        )

        return microgrid_info_from_proto(response.microgrid)

    async def list_components(  # noqa: DOC502 (raises ApiClientError indirectly)
        self,
        *,
        components: Iterable[ComponentId | Component] = (),
        categories: Iterable[ComponentCategory | int] = (),
    ) -> Iterable[ComponentTypes]:
        """Fetch all the components present in the local microgrid.

        Electrical components are a part of a microgrid's electrical infrastructure
        are can be connected to each other to form an electrical circuit, which can
        then be represented as a graph.

        If provided, the filters for component and categories have an `AND`
        relationship with one another, meaning that they are applied serially,
        but the elements within a single filter list have an `OR` relationship with
        each other.

        Example:
            If `ids = {1, 2, 3}`, and `categories = {ComponentCategory.INVERTER,
            ComponentCategory.BATTERY}`, then the results will consist of elements that
            have:

            * The IDs 1, `OR` 2, `OR` 3; `AND`
            * Are of the categories `ComponentCategory.INVERTER` `OR`
              `ComponentCategory.BATTERY`.

        If a filter list is empty, then that filter is not applied.

        Args:
            components: The components to fetch. See the method description for details.
            categories: The categories of the components to fetch. See the method
                description for details.

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

        Raises:
            ApiClientError: If the are any errors communicating with the Microgrid API,
                most likely a subclass of
                [GrpcError][frequenz.client.microgrid.GrpcError].
        """
        response = await client.call_stub_method(
            self,
            lambda: self.stub.ListElectricalComponents(
                microgrid_pb2.ListElectricalComponentsRequest(
                    electrical_component_ids=map(_get_component_id, components),
                    electrical_component_categories=map(
                        _get_category_value, categories
                    ),
                ),
                timeout=DEFAULT_GRPC_CALL_TIMEOUT,
            ),
            method_name="ListComponents",
        )

        return map(component_from_proto, response.electrical_components)

    async def list_connections(  # noqa: DOC502 (raises ApiClientError indirectly)
        self,
        *,
        sources: Iterable[ComponentId | Component] = (),
        destinations: Iterable[ComponentId | Component] = (),
    ) -> Iterable[ComponentConnection]:
        """Fetch all the connections present in the local microgrid.

        Electrical components are a part of a microgrid's electrical infrastructure
        are can be connected to each other to form an electrical circuit, which can
        then be represented as a graph.

        The direction of a connection is always away from the grid endpoint, i.e.
        aligned with the direction of positive current according to the passive sign
        convention: https://en.wikipedia.org/wiki/Passive_sign_convention

        The request may be filtered by `source`/`destination` component(s) of individual
        connections.  If provided, the `sources` and `destinations` filters have an
        `AND` relationship between each other, meaning that they are applied serially,
        but an `OR` relationship with other elements in the same list.

        Example:
            If `sources = {1, 2, 3}`, and `destinations = {4,
            5, 6}`, then the result should have all the connections where:

            * Each `source` component ID is either `1`, `2`, OR `3`; **AND**
            * Each `destination` component ID is either `4`, `5`, OR `6`.

        Args:
            sources: The component from which the connections originate.
            destinations: The component at which the connections terminate.

        Returns:
            Iterator whose elements are all the connections in the local microgrid.

        Raises:
            ApiClientError: If the are any errors communicating with the Microgrid API,
                most likely a subclass of
                [GrpcError][frequenz.client.microgrid.GrpcError].
        """
        response = await client.call_stub_method(
            self,
            lambda: self.stub.ListElectricalComponentConnections(
                microgrid_pb2.ListElectricalComponentConnectionsRequest(
                    source_electrical_component_ids=map(_get_component_id, sources),
                    destination_electrical_component_ids=map(
                        _get_component_id, destinations
                    ),
                ),
                timeout=DEFAULT_GRPC_CALL_TIMEOUT,
            ),
            method_name="ListConnections",
        )

        return (
            conn
            for conn in map(
                component_connection_from_proto,
                response.electrical_component_connections,
            )
            if conn is not None
        )

    # pylint: disable-next=fixme
    # TODO: Unifi set_component_power_active and set_component_power_reactive, or at
    #       least use a common implementation.
    #       Return an iterator or receiver with the streamed responses instead of
    #       returning just the first one
    async def set_component_power_active(  # noqa: DOC503
        self,
        component: ComponentId | Component,
        power: float,
        *,
        request_lifetime: timedelta | None = None,
        validate_arguments: bool = True,
    ) -> datetime | None:
        """Set the active power output of a component.

        The power output can be negative or positive, depending on whether the component
        is supposed to be discharging or charging, respectively.

        The power output is specified in watts.

        The return value is the timestamp until which the given power command will
        stay in effect. After this timestamp, the component's active power will be
        set to 0, if the API receives no further command to change it before then.
        By default, this timestamp will be set to the current time plus 60 seconds.

        Note:
            The target component may have a resolution of more than 1 W. E.g., an
            inverter may have a resolution of 88 W. In such cases, the magnitude of
            power will be floored to the nearest multiple of the resolution.

        Args:
            component: The component to set the output active power of.
            power: The output active power level, in watts. Negative values are for
                discharging, and positive values are for charging.
            request_lifetime: The duration, until which the request will stay in effect.
                This duration has to be between 10 seconds and 15 minutes (including
                both limits), otherwise the request will be rejected. It has
                a resolution of a second, so fractions of a second will be rounded for
                `timedelta` objects, and it is interpreted as seconds for `int` objects.
                If not provided, it usually defaults to 60 seconds.
            validate_arguments: Whether to validate the arguments before sending the
                request. If `True` a `ValueError` will be raised if an argument is
                invalid without even sending the request to the server, if `False`, the
                request will be sent without validation.

        Returns:
            The timestamp until which the given power command will stay in effect, or
                `None` if it was not provided by the server.

        Raises:
            ApiClientError: If there are any errors communicating with the Microgrid API,
                most likely a subclass of
                [GrpcError][frequenz.client.microgrid.GrpcError].
        """
        lifetime_seconds = _delta_to_seconds(request_lifetime)

        if validate_arguments:
            _validate_set_power_args(power=power, request_lifetime=lifetime_seconds)

        method_name = "SetElectricalComponentPower"
        if not self.is_connected:
            raise ClientNotConnected(server_url=self.server_url, operation=method_name)

        try:
            response = await anext(
                aiter(
                    self.stub.SetElectricalComponentPower(
                        microgrid_pb2.SetElectricalComponentPowerRequest(
                            electrical_component_id=_get_component_id(component),
                            power_type=microgrid_pb2.POWER_TYPE_ACTIVE,
                            power=power,
                            request_lifetime=lifetime_seconds,
                        ),
                        timeout=DEFAULT_GRPC_CALL_TIMEOUT,
                    )
                )
            )
        except AioRpcError as grpc_error:
            raise ApiClientError.from_grpc_error(
                server_url=self.server_url,
                operation=method_name,
                grpc_error=grpc_error,
            ) from grpc_error

        if response.HasField("valid_until_time"):
            return conversion.to_datetime(response.valid_until_time)

        return None

    async def set_component_power_reactive(  # noqa: DOC503
        self,
        component: ComponentId | Component,
        power: float,
        *,
        request_lifetime: timedelta | None = None,
        validate_arguments: bool = True,
    ) -> datetime | None:
        """Set the reactive power output of a component.

        We follow the polarity specified in the IEEE 1459-2010 standard
        definitions, where:

        - Positive reactive is inductive (current is lagging the voltage)
        - Negative reactive is capacitive (current is leading the voltage)

        The power output is specified in VAr.

        The return value is the timestamp until which the given power command will
        stay in effect. After this timestamp, the component's reactive power will
        be set to 0, if the API receives no further command to change it before
        then. By default, this timestamp will be set to the current time plus 60
        seconds.

        Note:
            The target component may have a resolution of more than 1 VAr. E.g., an
            inverter may have a resolution of 88 VAr. In such cases, the magnitude of
            power will be floored to the nearest multiple of the resolution.

        Args:
            component: The component to set the output reactive power of.
            power: The output reactive power level, in VAr. The standard of polarity is
                as per the IEEE 1459-2010 standard definitions: positive reactive is
                inductive (current is lagging the voltage); negative reactive is
                capacitive (current is leading the voltage).
            request_lifetime: The duration, until which the request will stay in effect.
                This duration has to be between 10 seconds and 15 minutes (including
                both limits), otherwise the request will be rejected. It has
                a resolution of a second, so fractions of a second will be rounded for
                `timedelta` objects, and it is interpreted as seconds for `int` objects.
                If not provided, it usually defaults to 60 seconds.
            validate_arguments: Whether to validate the arguments before sending the
                request. If `True` a `ValueError` will be raised if an argument is
                invalid without even sending the request to the server, if `False`, the
                request will be sent without validation.

        Returns:
            The timestamp until which the given power command will stay in effect, or
                `None` if it was not provided by the server.

        Raises:
            ApiClientError: If there are any errors communicating with the Microgrid API,
                most likely a subclass of
                [GrpcError][frequenz.client.microgrid.GrpcError].
        """
        lifetime_seconds = _delta_to_seconds(request_lifetime)

        if validate_arguments:
            _validate_set_power_args(power=power, request_lifetime=lifetime_seconds)

        method_name = "SetElectricalComponentPower"
        if not self.is_connected:
            raise ClientNotConnected(server_url=self.server_url, operation=method_name)

        try:
            response = await anext(
                aiter(
                    self.stub.SetElectricalComponentPower(
                        microgrid_pb2.SetElectricalComponentPowerRequest(
                            electrical_component_id=_get_component_id(component),
                            power_type=microgrid_pb2.POWER_TYPE_REACTIVE,
                            power=power,
                            request_lifetime=lifetime_seconds,
                        ),
                        timeout=DEFAULT_GRPC_CALL_TIMEOUT,
                    )
                )
            )
        except AioRpcError as grpc_error:
            raise ApiClientError.from_grpc_error(
                server_url=self.server_url,
                operation=method_name,
                grpc_error=grpc_error,
            ) from grpc_error

        if response.HasField("valid_until_time"):
            return conversion.to_datetime(response.valid_until_time)

        return None

    async def add_component_bounds(  # noqa: DOC502 (Raises ApiClientError indirectly)
        self,
        component: ComponentId | Component,
        target: Metric | int,
        bounds: Iterable[Bounds],
        *,
        validity: Validity | None = None,
    ) -> datetime | None:
        """Add inclusion bounds for a given metric of a given component.

        The bounds are used to define the acceptable range of values for a metric
        of a component. The added bounds are kept only temporarily, and removed
        automatically after some expiry time.

        Inclusion bounds give the range that the system will try to keep the
        metric within. If the metric goes outside of these bounds, the system will
        try to bring it back within the bounds.
        If the bounds for a metric are `[[lower_1, upper_1], [lower_2, upper_2]]`,
        then this metric's `value` needs to comply with the constraints `lower_1 <=
        value <= upper_1` OR `lower_2 <= value <= upper_2`.

        If multiple inclusion bounds have been provided for a metric, then the
        overlapping bounds are merged into a single bound, and non-overlapping
        bounds are kept separate.

        Example:
            If the bounds are [[0, 10], [5, 15], [20, 30]], then the resulting bounds
            will be [[0, 15], [20, 30]].

            The following diagram illustrates how bounds are applied:

            ```
              lower_1  upper_1
            <----|========|--------|========|-------->
                                lower_2  upper_2
            ```

            The bounds in this example are `[[lower_1, upper_1], [lower_2, upper_2]]`.

            ```
            ---- values here are considered out of range.
            ==== values here are considered within range.
            ```

        Note:
            For power metrics, regardless of the bounds, 0W is always allowed.

        Args:
            component: The component to add bounds to.
            target: The target metric whose bounds have to be added.
            bounds: The bounds to add to the target metric. Overlapping pairs of bounds
                are merged into a single pair of bounds, and non-overlapping ones are
                kept separated.
            validity: The duration for which the given bounds will stay in effect.
                If `None`, then the bounds will be removed after some default time
                decided by the server, typically 5 seconds.

                The duration for which the bounds are valid. If not provided, the
                bounds are considered to be valid indefinitely.

        Returns:
            The timestamp until which the given bounds will stay in effect, or `None` if
                if it was not provided by the server.

        Raises:
            ApiClientError: If there are any errors communicating with the Microgrid API,
                most likely a subclass of
                [GrpcError][frequenz.client.microgrid.GrpcError].
        """
        response = await client.call_stub_method(
            self,
            lambda: self.stub.AugmentElectricalComponentBounds(
                microgrid_pb2.AugmentElectricalComponentBoundsRequest(
                    electrical_component_id=_get_component_id(component),
                    target_metric=_get_metric_value(target),
                    bounds=(
                        bounds_pb2.Bounds(
                            lower=bound.lower,
                            upper=bound.upper,
                        )
                        for bound in bounds
                    ),
                    request_lifetime=validity.value if validity else None,
                ),
                timeout=DEFAULT_GRPC_CALL_TIMEOUT,
            ),
            method_name="AddComponentBounds",
        )

        if response.HasField("valid_until_time"):
            return conversion.to_datetime(response.valid_until_time)

        return None

    # noqa: DOC502 (Raises ApiClientError indirectly)
    def receive_component_data_samples_stream(
        self,
        component: ComponentId | Component,
        metrics: Iterable[Metric | int],
        *,
        buffer_size: int = 50,
    ) -> Receiver[ComponentDataSamples]:
        """Stream data samples from a component.

        At least one metric must be specified. If no metric is specified, then the
        stream will raise an error.

        Warning:
            Components may not support all metrics. If a component does not support
            a given metric, then the returned data stream will not contain that metric.

            There is no way to tell if a metric is not being received because the
            component does not support it or because there is a transient issue when
            retrieving the metric from the component.

            The supported metrics by a component can even change with time, for example,
            if a component is updated with new firmware.

        Args:
            component: The component to stream data from.
            metrics: List of metrics to return. Only the specified metrics will be
                returned.
            buffer_size: The maximum number of messages to buffer in the returned
                receiver. After this limit is reached, the oldest messages will be
                dropped.

        Returns:
            The data stream from the component.
        """
        component_id = _get_component_id(component)
        metrics_set = frozenset([_get_metric_value(m) for m in metrics])
        key = f"{component_id}-{hash(metrics_set)}"
        broadcaster = self._component_data_broadcasters.get(key)
        if broadcaster is None:
            client_id = hex(id(self))[2:]
            stream_name = f"microgrid-client-{client_id}-component-data-{key}"
            # Alias to avoid too long lines linter errors
            # pylint: disable-next=invalid-name
            Request = microgrid_pb2.ReceiveElectricalComponentTelemetryStreamRequest
            broadcaster = streaming.GrpcStreamBroadcaster(
                stream_name,
                lambda: aiter(
                    self.stub.ReceiveElectricalComponentTelemetryStream(
                        Request(
                            electrical_component_id=_get_component_id(component),
                            filter=Request.ComponentTelemetryStreamFilter(
                                metrics=metrics_set
                            ),
                        ),
                    )
                ),
                lambda msg: component_data_samples_from_proto(msg.telemetry),
                retry_strategy=self._retry_strategy,
            )
            self._component_data_broadcasters[key] = broadcaster
        return broadcaster.new_receiver(maxsize=buffer_size)
Attributes¤
channel property ¤
channel: Channel

The underlying gRPC channel used to communicate with the server.

Warning

This channel is provided as a last resort for advanced users. It is not recommended to use this property directly unless you know what you are doing and you don't care about being tied to a specific gRPC library.

RAISES DESCRIPTION
ClientNotConnected

If the client is not connected to the server.
channel_defaults property ¤
channel_defaults: ChannelOptions

The default options for the gRPC channel.

is_connected property ¤
is_connected: bool

Whether the client is connected to the server.

server_url property ¤
server_url: str

The URL of the server.

stub property ¤
stub: MicrogridAsyncStub

The gRPC stub for the API.

Functions¤
__aenter__ async ¤
__aenter__() -> Self

Enter a context manager.

Source code in frequenz/client/base/client.py 
294
295
296
297
async def __aenter__(self) -> Self:
    """Enter a context manager."""
    self.connect()
    return self
__aexit__ async ¤
__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: Any | None,
) -> bool | None

Close the gRPC channel and stop all broadcasters.

Source code in frequenz/client/microgrid/_client.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
@override
async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: Any | None,
) -> bool | None:
    """Close the gRPC channel and stop all broadcasters."""
    exceptions = list(
        exc
        for exc in await asyncio.gather(
            *(
                broadcaster.stop()
                for broadcaster in itertools.chain(
                    self._component_data_broadcasters.values(),
                    self._sensor_data_broadcasters.values(),
                )
            ),
            return_exceptions=True,
        )
        if isinstance(exc, BaseException)
    )
    self._component_data_broadcasters.clear()
    self._sensor_data_broadcasters.clear()

    result = None
    try:
        result = await super().__aexit__(exc_type, exc_val, exc_tb)
    except Exception as exc:  # pylint: disable=broad-except
        exceptions.append(exc)
    if exceptions:
        raise BaseExceptionGroup(
            "Error while disconnecting from the microgrid API", exceptions
        )
    return result
__init__ ¤
__init__(
    server_url: str,
    *,
    channel_defaults: ChannelOptions = DEFAULT_CHANNEL_OPTIONS,
    connect: bool = True,
    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_defaults

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

TYPE: ChannelOptions DEFAULT: DEFAULT_CHANNEL_OPTIONS

connect

Whether to connect to the server as soon as a client instance is created. If False, the client will not connect to the server until connect() is called.

TYPE: bool DEFAULT: True

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 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def __init__(
    self,
    server_url: str,
    *,
    channel_defaults: channel.ChannelOptions = DEFAULT_CHANNEL_OPTIONS,
    connect: bool = True,
    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_defaults: The default options use to create the channel when not
            specified in the URL.
        connect: Whether to connect to the server as soon as a client instance is
            created. If `False`, the client will not connect to the server until
            [connect()][frequenz.client.base.client.BaseApiClient.connect] is
            called.
        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.
    """
    super().__init__(
        server_url,
        microgrid_pb2_grpc.MicrogridStub,
        connect=connect,
        channel_defaults=channel_defaults,
    )
    self._component_data_broadcasters: dict[
        str,
        streaming.GrpcStreamBroadcaster[
            microgrid_pb2.ReceiveElectricalComponentTelemetryStreamResponse,
            ComponentDataSamples,
        ],
    ] = {}
    self._sensor_data_broadcasters: dict[
        str,
        streaming.GrpcStreamBroadcaster[Any, Any],
    ] = {}
    self._retry_strategy = retry_strategy
add_component_bounds async ¤
add_component_bounds(
    component: ComponentId | Component,
    target: Metric | int,
    bounds: Iterable[Bounds],
    *,
    validity: Validity | None = None
) -> datetime | None

Add inclusion bounds for a given metric of a given component.

The bounds are used to define the acceptable range of values for a metric of a component. The added bounds are kept only temporarily, and removed automatically after some expiry time.

Inclusion bounds give the range that the system will try to keep the metric within. If the metric goes outside of these bounds, the system will try to bring it back within the bounds. If the bounds for a metric are [[lower_1, upper_1], [lower_2, upper_2]], then this metric's value needs to comply with the constraints lower_1 <= value <= upper_1 OR lower_2 <= value <= upper_2.

If multiple inclusion bounds have been provided for a metric, then the overlapping bounds are merged into a single bound, and non-overlapping bounds are kept separate.

Example

If the bounds are [[0, 10], [5, 15], [20, 30]], then the resulting bounds will be [[0, 15], [20, 30]].

The following diagram illustrates how bounds are applied:

  lower_1  upper_1
<----|========|--------|========|-------->
                    lower_2  upper_2

The bounds in this example are [[lower_1, upper_1], [lower_2, upper_2]].

---- values here are considered out of range.
==== values here are considered within range.
Note

For power metrics, regardless of the bounds, 0W is always allowed.

PARAMETER DESCRIPTION
component

The component to add bounds to.

TYPE: ComponentId | Component

target

The target metric whose bounds have to be added.

TYPE: Metric | int

bounds

The bounds to add to the target metric. Overlapping pairs of bounds are merged into a single pair of bounds, and non-overlapping ones are kept separated.

TYPE: Iterable[Bounds]

validity

The duration for which the given bounds will stay in effect. If None, then the bounds will be removed after some default time decided by the server, typically 5 seconds.

The duration for which the bounds are valid. If not provided, the bounds are considered to be valid indefinitely.

TYPE: Validity | None DEFAULT: None

RETURNS DESCRIPTION
datetime | None

The timestamp until which the given bounds will stay in effect, or None if if it was not provided by the server.
RAISES DESCRIPTION
ApiClientError

If there are any errors communicating with the Microgrid API, most likely a subclass of GrpcError.
Source code in frequenz/client/microgrid/_client.py 
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
async def add_component_bounds(  # noqa: DOC502 (Raises ApiClientError indirectly)
    self,
    component: ComponentId | Component,
    target: Metric | int,
    bounds: Iterable[Bounds],
    *,
    validity: Validity | None = None,
) -> datetime | None:
    """Add inclusion bounds for a given metric of a given component.

    The bounds are used to define the acceptable range of values for a metric
    of a component. The added bounds are kept only temporarily, and removed
    automatically after some expiry time.

    Inclusion bounds give the range that the system will try to keep the
    metric within. If the metric goes outside of these bounds, the system will
    try to bring it back within the bounds.
    If the bounds for a metric are `[[lower_1, upper_1], [lower_2, upper_2]]`,
    then this metric's `value` needs to comply with the constraints `lower_1 <=
    value <= upper_1` OR `lower_2 <= value <= upper_2`.

    If multiple inclusion bounds have been provided for a metric, then the
    overlapping bounds are merged into a single bound, and non-overlapping
    bounds are kept separate.

    Example:
        If the bounds are [[0, 10], [5, 15], [20, 30]], then the resulting bounds
        will be [[0, 15], [20, 30]].

        The following diagram illustrates how bounds are applied:

        ```
          lower_1  upper_1
        <----|========|--------|========|-------->
                            lower_2  upper_2
        ```

        The bounds in this example are `[[lower_1, upper_1], [lower_2, upper_2]]`.

        ```
        ---- values here are considered out of range.
        ==== values here are considered within range.
        ```

    Note:
        For power metrics, regardless of the bounds, 0W is always allowed.

    Args:
        component: The component to add bounds to.
        target: The target metric whose bounds have to be added.
        bounds: The bounds to add to the target metric. Overlapping pairs of bounds
            are merged into a single pair of bounds, and non-overlapping ones are
            kept separated.
        validity: The duration for which the given bounds will stay in effect.
            If `None`, then the bounds will be removed after some default time
            decided by the server, typically 5 seconds.

            The duration for which the bounds are valid. If not provided, the
            bounds are considered to be valid indefinitely.

    Returns:
        The timestamp until which the given bounds will stay in effect, or `None` if
            if it was not provided by the server.

    Raises:
        ApiClientError: If there are any errors communicating with the Microgrid API,
            most likely a subclass of
            [GrpcError][frequenz.client.microgrid.GrpcError].
    """
    response = await client.call_stub_method(
        self,
        lambda: self.stub.AugmentElectricalComponentBounds(
            microgrid_pb2.AugmentElectricalComponentBoundsRequest(
                electrical_component_id=_get_component_id(component),
                target_metric=_get_metric_value(target),
                bounds=(
                    bounds_pb2.Bounds(
                        lower=bound.lower,
                        upper=bound.upper,
                    )
                    for bound in bounds
                ),
                request_lifetime=validity.value if validity else None,
            ),
            timeout=DEFAULT_GRPC_CALL_TIMEOUT,
        ),
        method_name="AddComponentBounds",
    )

    if response.HasField("valid_until_time"):
        return conversion.to_datetime(response.valid_until_time)

    return None
connect ¤
connect(
    server_url: str | None = None,
    *,
    auth_key: str | None | EllipsisType = ...,
    sign_secret: str | None | EllipsisType = ...
) -> None

Connect to the server, possibly using a new URL.

If the client is already connected and the URL is the same as the previous URL, this method does nothing. If you want to force a reconnection, you can call disconnect() first.

PARAMETER DESCRIPTION
server_url

The URL of the server to connect to. If not provided, the previously used URL is used.

TYPE: str | None DEFAULT: None

auth_key

The API key to use when connecting to the service. If an Ellipsis is provided, the previously used auth_key is used.

TYPE: str | None | EllipsisType DEFAULT: ...

sign_secret

The secret to use when creating message HMAC. If an Ellipsis is provided,

TYPE: str | None | EllipsisType DEFAULT: ...

Source code in frequenz/client/base/client.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
def connect(
    self,
    server_url: str | None = None,
    *,
    auth_key: str | None | EllipsisType = ...,
    sign_secret: str | None | EllipsisType = ...,
) -> None:
    """Connect to the server, possibly using a new URL.

    If the client is already connected and the URL is the same as the previous URL,
    this method does nothing. If you want to force a reconnection, you can call
    [disconnect()][frequenz.client.base.client.BaseApiClient.disconnect] first.

    Args:
        server_url: The URL of the server to connect to. If not provided, the
            previously used URL is used.
        auth_key: The API key to use when connecting to the service. If an Ellipsis
            is provided, the previously used auth_key is used.
        sign_secret: The secret to use when creating message HMAC. If an Ellipsis is
            provided,
    """
    reconnect = False
    if server_url is not None and server_url != self._server_url:  # URL changed
        self._server_url = server_url
        reconnect = True
    if auth_key is not ... and auth_key != self._auth_key:
        self._auth_key = auth_key
        reconnect = True
    if sign_secret is not ... and sign_secret != self._sign_secret:
        self._sign_secret = sign_secret
        reconnect = True
    if self.is_connected and not reconnect:  # Desired connection already exists
        return

    interceptors: list[ClientInterceptor] = []
    if self._auth_key is not None:
        interceptors += [
            AuthenticationInterceptorUnaryUnary(self._auth_key),  # type: ignore [list-item]
            AuthenticationInterceptorUnaryStream(self._auth_key),  # type: ignore [list-item]
        ]
    if self._sign_secret is not None:
        interceptors += [
            SigningInterceptorUnaryUnary(self._sign_secret),  # type: ignore [list-item]
            SigningInterceptorUnaryStream(self._sign_secret),  # type: ignore [list-item]
        ]

    self._channel = parse_grpc_uri(
        self._server_url,
        interceptors,
        defaults=self._channel_defaults,
    )
    self._stub = self._create_stub(self._channel)
disconnect async ¤
disconnect() -> None

Disconnect from the server.

If the client is not connected, this method does nothing.

Source code in frequenz/client/base/client.py 
287
288
289
290
291
292
async def disconnect(self) -> None:
    """Disconnect from the server.

    If the client is not connected, this method does nothing.
    """
    await self.__aexit__(None, None, None)
get_microgrid_info async ¤
get_microgrid_info() -> MicrogridInfo

Retrieve information about the local microgrid.

This consists of information about the overall microgrid, for example, the microgrid ID and its location. It does not include information about the electrical components or sensors in the microgrid.

RETURNS DESCRIPTION
MicrogridInfo

The information about the local microgrid.
RAISES DESCRIPTION
ApiClientError

If there are any errors communicating with the Microgrid API, most likely a subclass of GrpcError.
Source code in frequenz/client/microgrid/_client.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
async def get_microgrid_info(  # noqa: DOC502 (raises ApiClientError indirectly)
    self,
) -> MicrogridInfo:
    """Retrieve information about the local microgrid.

    This consists of information about the overall microgrid, for example, the
    microgrid ID and its location.  It does not include information about the
    electrical components or sensors in the microgrid.

    Returns:
        The information about the local microgrid.

    Raises:
        ApiClientError: If there are any errors communicating with the Microgrid API,
            most likely a subclass of
            [GrpcError][frequenz.client.microgrid.GrpcError].
    """
    response = await client.call_stub_method(
        self,
        lambda: self.stub.GetMicrogrid(Empty(), timeout=DEFAULT_GRPC_CALL_TIMEOUT),
        method_name="GetMicrogridMetadata",
    )

    return microgrid_info_from_proto(response.microgrid)
list_components async ¤
list_components(
    *,
    components: Iterable[ComponentId | Component] = (),
    categories: Iterable[ComponentCategory | int] = ()
) -> Iterable[ComponentTypes]

Fetch all the components present in the local microgrid.

Electrical components are a part of a microgrid's electrical infrastructure are can be connected to each other to form an electrical circuit, which can then be represented as a graph.

If provided, the filters for component and categories have an AND relationship with one another, meaning that they are applied serially, but the elements within a single filter list have an OR relationship with each other.

Example

If ids = {1, 2, 3}, and categories = {ComponentCategory.INVERTER, ComponentCategory.BATTERY}, then the results will consist of elements that have:

  • The IDs 1, OR 2, OR 3; AND
  • Are of the categories ComponentCategory.INVERTER OR ComponentCategory.BATTERY.

If a filter list is empty, then that filter is not applied.

PARAMETER DESCRIPTION
components

The components to fetch. See the method description for details.

TYPE: Iterable[ComponentId | Component] DEFAULT: ()

categories

The categories of the components to fetch. See the method description for details.

TYPE: Iterable[ComponentCategory | int] DEFAULT: ()

RETURNS DESCRIPTION
Iterable[ComponentTypes]

Iterator whose elements are all the components in the local 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 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
async def list_components(  # noqa: DOC502 (raises ApiClientError indirectly)
    self,
    *,
    components: Iterable[ComponentId | Component] = (),
    categories: Iterable[ComponentCategory | int] = (),
) -> Iterable[ComponentTypes]:
    """Fetch all the components present in the local microgrid.

    Electrical components are a part of a microgrid's electrical infrastructure
    are can be connected to each other to form an electrical circuit, which can
    then be represented as a graph.

    If provided, the filters for component and categories have an `AND`
    relationship with one another, meaning that they are applied serially,
    but the elements within a single filter list have an `OR` relationship with
    each other.

    Example:
        If `ids = {1, 2, 3}`, and `categories = {ComponentCategory.INVERTER,
        ComponentCategory.BATTERY}`, then the results will consist of elements that
        have:

        * The IDs 1, `OR` 2, `OR` 3; `AND`
        * Are of the categories `ComponentCategory.INVERTER` `OR`
          `ComponentCategory.BATTERY`.

    If a filter list is empty, then that filter is not applied.

    Args:
        components: The components to fetch. See the method description for details.
        categories: The categories of the components to fetch. See the method
            description for details.

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

    Raises:
        ApiClientError: If the are any errors communicating with the Microgrid API,
            most likely a subclass of
            [GrpcError][frequenz.client.microgrid.GrpcError].
    """
    response = await client.call_stub_method(
        self,
        lambda: self.stub.ListElectricalComponents(
            microgrid_pb2.ListElectricalComponentsRequest(
                electrical_component_ids=map(_get_component_id, components),
                electrical_component_categories=map(
                    _get_category_value, categories
                ),
            ),
            timeout=DEFAULT_GRPC_CALL_TIMEOUT,
        ),
        method_name="ListComponents",
    )

    return map(component_from_proto, response.electrical_components)
list_connections async ¤
list_connections(
    *,
    sources: Iterable[ComponentId | Component] = (),
    destinations: Iterable[ComponentId | Component] = ()
) -> Iterable[ComponentConnection]

Fetch all the connections present in the local microgrid.

Electrical components are a part of a microgrid's electrical infrastructure are can be connected to each other to form an electrical circuit, which can then be represented as a graph.

The direction of a connection is always away from the grid endpoint, i.e. aligned with the direction of positive current according to the passive sign convention: https://en.wikipedia.org/wiki/Passive_sign_convention

The request may be filtered by source/destination component(s) of individual connections. If provided, the sources and destinations filters have an AND relationship between each other, meaning that they are applied serially, but an OR relationship with other elements in the same list.

Example

If sources = {1, 2, 3}, and destinations = {4, 5, 6}, then the result should have all the connections where:

  • Each source component ID is either 1, 2, OR 3; AND
  • Each destination component ID is either 4, 5, OR 6.
PARAMETER DESCRIPTION
sources

The component from which the connections originate.

TYPE: Iterable[ComponentId | Component] DEFAULT: ()

destinations

The component at which the connections terminate.

TYPE: Iterable[ComponentId | Component] DEFAULT: ()

RETURNS DESCRIPTION
Iterable[ComponentConnection]

Iterator whose elements are all the connections in the local 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 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
async def list_connections(  # noqa: DOC502 (raises ApiClientError indirectly)
    self,
    *,
    sources: Iterable[ComponentId | Component] = (),
    destinations: Iterable[ComponentId | Component] = (),
) -> Iterable[ComponentConnection]:
    """Fetch all the connections present in the local microgrid.

    Electrical components are a part of a microgrid's electrical infrastructure
    are can be connected to each other to form an electrical circuit, which can
    then be represented as a graph.

    The direction of a connection is always away from the grid endpoint, i.e.
    aligned with the direction of positive current according to the passive sign
    convention: https://en.wikipedia.org/wiki/Passive_sign_convention

    The request may be filtered by `source`/`destination` component(s) of individual
    connections.  If provided, the `sources` and `destinations` filters have an
    `AND` relationship between each other, meaning that they are applied serially,
    but an `OR` relationship with other elements in the same list.

    Example:
        If `sources = {1, 2, 3}`, and `destinations = {4,
        5, 6}`, then the result should have all the connections where:

        * Each `source` component ID is either `1`, `2`, OR `3`; **AND**
        * Each `destination` component ID is either `4`, `5`, OR `6`.

    Args:
        sources: The component from which the connections originate.
        destinations: The component at which the connections terminate.

    Returns:
        Iterator whose elements are all the connections in the local microgrid.

    Raises:
        ApiClientError: If the are any errors communicating with the Microgrid API,
            most likely a subclass of
            [GrpcError][frequenz.client.microgrid.GrpcError].
    """
    response = await client.call_stub_method(
        self,
        lambda: self.stub.ListElectricalComponentConnections(
            microgrid_pb2.ListElectricalComponentConnectionsRequest(
                source_electrical_component_ids=map(_get_component_id, sources),
                destination_electrical_component_ids=map(
                    _get_component_id, destinations
                ),
            ),
            timeout=DEFAULT_GRPC_CALL_TIMEOUT,
        ),
        method_name="ListConnections",
    )

    return (
        conn
        for conn in map(
            component_connection_from_proto,
            response.electrical_component_connections,
        )
        if conn is not None
    )
receive_component_data_samples_stream ¤
receive_component_data_samples_stream(
    component: ComponentId | Component,
    metrics: Iterable[Metric | int],
    *,
    buffer_size: int = 50
) -> Receiver[ComponentDataSamples]

Stream data samples from a component.

At least one metric must be specified. If no metric is specified, then the stream will raise an error.

Warning

Components may not support all metrics. If a component does not support a given metric, then the returned data stream will not contain that metric.

There is no way to tell if a metric is not being received because the component does not support it or because there is a transient issue when retrieving the metric from the component.

The supported metrics by a component can even change with time, for example, if a component is updated with new firmware.

PARAMETER DESCRIPTION
component

The component to stream data from.

TYPE: ComponentId | Component

metrics

List of metrics to return. Only the specified metrics will be returned.

TYPE: Iterable[Metric | int]

buffer_size

The maximum number of messages to buffer in the returned receiver. After this limit is reached, the oldest messages will be dropped.

TYPE: int DEFAULT: 50

RETURNS DESCRIPTION
Receiver[ComponentDataSamples]

The data stream from the component.
Source code in frequenz/client/microgrid/_client.py 
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
def receive_component_data_samples_stream(
    self,
    component: ComponentId | Component,
    metrics: Iterable[Metric | int],
    *,
    buffer_size: int = 50,
) -> Receiver[ComponentDataSamples]:
    """Stream data samples from a component.

    At least one metric must be specified. If no metric is specified, then the
    stream will raise an error.

    Warning:
        Components may not support all metrics. If a component does not support
        a given metric, then the returned data stream will not contain that metric.

        There is no way to tell if a metric is not being received because the
        component does not support it or because there is a transient issue when
        retrieving the metric from the component.

        The supported metrics by a component can even change with time, for example,
        if a component is updated with new firmware.

    Args:
        component: The component to stream data from.
        metrics: List of metrics to return. Only the specified metrics will be
            returned.
        buffer_size: The maximum number of messages to buffer in the returned
            receiver. After this limit is reached, the oldest messages will be
            dropped.

    Returns:
        The data stream from the component.
    """
    component_id = _get_component_id(component)
    metrics_set = frozenset([_get_metric_value(m) for m in metrics])
    key = f"{component_id}-{hash(metrics_set)}"
    broadcaster = self._component_data_broadcasters.get(key)
    if broadcaster is None:
        client_id = hex(id(self))[2:]
        stream_name = f"microgrid-client-{client_id}-component-data-{key}"
        # Alias to avoid too long lines linter errors
        # pylint: disable-next=invalid-name
        Request = microgrid_pb2.ReceiveElectricalComponentTelemetryStreamRequest
        broadcaster = streaming.GrpcStreamBroadcaster(
            stream_name,
            lambda: aiter(
                self.stub.ReceiveElectricalComponentTelemetryStream(
                    Request(
                        electrical_component_id=_get_component_id(component),
                        filter=Request.ComponentTelemetryStreamFilter(
                            metrics=metrics_set
                        ),
                    ),
                )
            ),
            lambda msg: component_data_samples_from_proto(msg.telemetry),
            retry_strategy=self._retry_strategy,
        )
        self._component_data_broadcasters[key] = broadcaster
    return broadcaster.new_receiver(maxsize=buffer_size)
set_component_power_active async ¤
set_component_power_active(
    component: ComponentId | Component,
    power: float,
    *,
    request_lifetime: timedelta | None = None,
    validate_arguments: bool = True
) -> datetime | None

Set the active power output of a component.

The power output can be negative or positive, depending on whether the component is supposed to be discharging or charging, respectively.

The power output is specified in watts.

The return value is the timestamp until which the given power command will stay in effect. After this timestamp, the component's active power will be set to 0, if the API receives no further command to change it before then. By default, this timestamp will be set to the current time plus 60 seconds.

Note

The target component may have a resolution of more than 1 W. E.g., an inverter may have a resolution of 88 W. In such cases, the magnitude of power will be floored to the nearest multiple of the resolution.

PARAMETER DESCRIPTION
component

The component to set the output active power of.

TYPE: ComponentId | Component

power

The output active power level, in watts. Negative values are for discharging, and positive values are for charging.

TYPE: float

request_lifetime

The duration, until which the request will stay in effect. This duration has to be between 10 seconds and 15 minutes (including both limits), otherwise the request will be rejected. It has a resolution of a second, so fractions of a second will be rounded for timedelta objects, and it is interpreted as seconds for int objects. If not provided, it usually defaults to 60 seconds.

TYPE: timedelta | None DEFAULT: None

validate_arguments

Whether to validate the arguments before sending the request. If True a ValueError will be raised if an argument is invalid without even sending the request to the server, if False, the request will be sent without validation.

TYPE: bool DEFAULT: True

RETURNS DESCRIPTION
datetime | None

The timestamp until which the given power command will stay in effect, or None if it was not provided by the server.
RAISES DESCRIPTION
ApiClientError

If there are any errors communicating with the Microgrid API, most likely a subclass of GrpcError.
Source code in frequenz/client/microgrid/_client.py 
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
async def set_component_power_active(  # noqa: DOC503
    self,
    component: ComponentId | Component,
    power: float,
    *,
    request_lifetime: timedelta | None = None,
    validate_arguments: bool = True,
) -> datetime | None:
    """Set the active power output of a component.

    The power output can be negative or positive, depending on whether the component
    is supposed to be discharging or charging, respectively.

    The power output is specified in watts.

    The return value is the timestamp until which the given power command will
    stay in effect. After this timestamp, the component's active power will be
    set to 0, if the API receives no further command to change it before then.
    By default, this timestamp will be set to the current time plus 60 seconds.

    Note:
        The target component may have a resolution of more than 1 W. E.g., an
        inverter may have a resolution of 88 W. In such cases, the magnitude of
        power will be floored to the nearest multiple of the resolution.

    Args:
        component: The component to set the output active power of.
        power: The output active power level, in watts. Negative values are for
            discharging, and positive values are for charging.
        request_lifetime: The duration, until which the request will stay in effect.
            This duration has to be between 10 seconds and 15 minutes (including
            both limits), otherwise the request will be rejected. It has
            a resolution of a second, so fractions of a second will be rounded for
            `timedelta` objects, and it is interpreted as seconds for `int` objects.
            If not provided, it usually defaults to 60 seconds.
        validate_arguments: Whether to validate the arguments before sending the
            request. If `True` a `ValueError` will be raised if an argument is
            invalid without even sending the request to the server, if `False`, the
            request will be sent without validation.

    Returns:
        The timestamp until which the given power command will stay in effect, or
            `None` if it was not provided by the server.

    Raises:
        ApiClientError: If there are any errors communicating with the Microgrid API,
            most likely a subclass of
            [GrpcError][frequenz.client.microgrid.GrpcError].
    """
    lifetime_seconds = _delta_to_seconds(request_lifetime)

    if validate_arguments:
        _validate_set_power_args(power=power, request_lifetime=lifetime_seconds)

    method_name = "SetElectricalComponentPower"
    if not self.is_connected:
        raise ClientNotConnected(server_url=self.server_url, operation=method_name)

    try:
        response = await anext(
            aiter(
                self.stub.SetElectricalComponentPower(
                    microgrid_pb2.SetElectricalComponentPowerRequest(
                        electrical_component_id=_get_component_id(component),
                        power_type=microgrid_pb2.POWER_TYPE_ACTIVE,
                        power=power,
                        request_lifetime=lifetime_seconds,
                    ),
                    timeout=DEFAULT_GRPC_CALL_TIMEOUT,
                )
            )
        )
    except AioRpcError as grpc_error:
        raise ApiClientError.from_grpc_error(
            server_url=self.server_url,
            operation=method_name,
            grpc_error=grpc_error,
        ) from grpc_error

    if response.HasField("valid_until_time"):
        return conversion.to_datetime(response.valid_until_time)

    return None
set_component_power_reactive async ¤
set_component_power_reactive(
    component: ComponentId | Component,
    power: float,
    *,
    request_lifetime: timedelta | None = None,
    validate_arguments: bool = True
) -> datetime | None

Set the reactive power output of a component.

We follow the polarity specified in the IEEE 1459-2010 standard definitions, where:

  • Positive reactive is inductive (current is lagging the voltage)
  • Negative reactive is capacitive (current is leading the voltage)

The power output is specified in VAr.

The return value is the timestamp until which the given power command will stay in effect. After this timestamp, the component's reactive power will be set to 0, if the API receives no further command to change it before then. By default, this timestamp will be set to the current time plus 60 seconds.

Note

The target component may have a resolution of more than 1 VAr. E.g., an inverter may have a resolution of 88 VAr. In such cases, the magnitude of power will be floored to the nearest multiple of the resolution.

PARAMETER DESCRIPTION
component

The component to set the output reactive power of.

TYPE: ComponentId | Component

power

The output reactive power level, in VAr. The standard of polarity is as per the IEEE 1459-2010 standard definitions: positive reactive is inductive (current is lagging the voltage); negative reactive is capacitive (current is leading the voltage).

TYPE: float

request_lifetime

The duration, until which the request will stay in effect. This duration has to be between 10 seconds and 15 minutes (including both limits), otherwise the request will be rejected. It has a resolution of a second, so fractions of a second will be rounded for timedelta objects, and it is interpreted as seconds for int objects. If not provided, it usually defaults to 60 seconds.

TYPE: timedelta | None DEFAULT: None

validate_arguments

Whether to validate the arguments before sending the request. If True a ValueError will be raised if an argument is invalid without even sending the request to the server, if False, the request will be sent without validation.

TYPE: bool DEFAULT: True

RETURNS DESCRIPTION
datetime | None

The timestamp until which the given power command will stay in effect, or None if it was not provided by the server.
RAISES DESCRIPTION
ApiClientError

If there are any errors communicating with the Microgrid API, most likely a subclass of GrpcError.
Source code in frequenz/client/microgrid/_client.py 
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
async def set_component_power_reactive(  # noqa: DOC503
    self,
    component: ComponentId | Component,
    power: float,
    *,
    request_lifetime: timedelta | None = None,
    validate_arguments: bool = True,
) -> datetime | None:
    """Set the reactive power output of a component.

    We follow the polarity specified in the IEEE 1459-2010 standard
    definitions, where:

    - Positive reactive is inductive (current is lagging the voltage)
    - Negative reactive is capacitive (current is leading the voltage)

    The power output is specified in VAr.

    The return value is the timestamp until which the given power command will
    stay in effect. After this timestamp, the component's reactive power will
    be set to 0, if the API receives no further command to change it before
    then. By default, this timestamp will be set to the current time plus 60
    seconds.

    Note:
        The target component may have a resolution of more than 1 VAr. E.g., an
        inverter may have a resolution of 88 VAr. In such cases, the magnitude of
        power will be floored to the nearest multiple of the resolution.

    Args:
        component: The component to set the output reactive power of.
        power: The output reactive power level, in VAr. The standard of polarity is
            as per the IEEE 1459-2010 standard definitions: positive reactive is
            inductive (current is lagging the voltage); negative reactive is
            capacitive (current is leading the voltage).
        request_lifetime: The duration, until which the request will stay in effect.
            This duration has to be between 10 seconds and 15 minutes (including
            both limits), otherwise the request will be rejected. It has
            a resolution of a second, so fractions of a second will be rounded for
            `timedelta` objects, and it is interpreted as seconds for `int` objects.
            If not provided, it usually defaults to 60 seconds.
        validate_arguments: Whether to validate the arguments before sending the
            request. If `True` a `ValueError` will be raised if an argument is
            invalid without even sending the request to the server, if `False`, the
            request will be sent without validation.

    Returns:
        The timestamp until which the given power command will stay in effect, or
            `None` if it was not provided by the server.

    Raises:
        ApiClientError: If there are any errors communicating with the Microgrid API,
            most likely a subclass of
            [GrpcError][frequenz.client.microgrid.GrpcError].
    """
    lifetime_seconds = _delta_to_seconds(request_lifetime)

    if validate_arguments:
        _validate_set_power_args(power=power, request_lifetime=lifetime_seconds)

    method_name = "SetElectricalComponentPower"
    if not self.is_connected:
        raise ClientNotConnected(server_url=self.server_url, operation=method_name)

    try:
        response = await anext(
            aiter(
                self.stub.SetElectricalComponentPower(
                    microgrid_pb2.SetElectricalComponentPowerRequest(
                        electrical_component_id=_get_component_id(component),
                        power_type=microgrid_pb2.POWER_TYPE_REACTIVE,
                        power=power,
                        request_lifetime=lifetime_seconds,
                    ),
                    timeout=DEFAULT_GRPC_CALL_TIMEOUT,
                )
            )
        )
    except AioRpcError as grpc_error:
        raise ApiClientError.from_grpc_error(
            server_url=self.server_url,
            operation=method_name,
            grpc_error=grpc_error,
        ) from grpc_error

    if response.HasField("valid_until_time"):
        return conversion.to_datetime(response.valid_until_time)

    return None

frequenz.client.microgrid.MicrogridInfo dataclass ¤

A localized grouping of electricity generation, energy storage, and loads.

A microgrid is a localized grouping of electricity generation, energy storage, and loads that normally operates connected to a traditional centralized grid.

Each microgrid has a unique identifier and is associated with an enterprise account.

A key feature is that it has a physical location and is situated in a delivery area.

Key Concepts
  • Physical Location: Geographical coordinates specify the exact physical location of the microgrid.
  • Delivery Area: Each microgrid is part of a broader delivery area, which is crucial for energy trading and compliance.
Source code in frequenz/client/microgrid/_microgrid_info.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@dataclass(frozen=True, kw_only=True)
class MicrogridInfo:
    """A localized grouping of electricity generation, energy storage, and loads.

    A microgrid is a localized grouping of electricity generation, energy storage, and
    loads that normally operates connected to a traditional centralized grid.

    Each microgrid has a unique identifier and is associated with an enterprise account.

    A key feature is that it has a physical location and is situated in a delivery area.

    Note: Key Concepts
        - Physical Location: Geographical coordinates specify the exact physical
          location of the microgrid.
        - Delivery Area: Each microgrid is part of a broader delivery area, which is
          crucial for energy trading and compliance.
    """

    id: MicrogridId
    """The unique identifier of the microgrid."""

    enterprise_id: EnterpriseId
    """The unique identifier linking this microgrid to its parent enterprise account."""

    name: str | None
    """Name of the microgrid."""

    delivery_area: DeliveryArea | None
    """The delivery area where the microgrid is located, as identified by a specific code."""

    location: Location | None
    """Physical location of the microgrid, in geographical co-ordinates."""

    status: MicrogridStatus | int
    """The current status of the microgrid."""

    create_timestamp: datetime.datetime
    """The UTC timestamp indicating when the microgrid was initially created."""

    @cached_property
    def is_active(self) -> bool:
        """Whether the microgrid is active."""
        if self.status is MicrogridStatus.UNSPECIFIED:
            # Because this is a cached property, the warning will only be logged once.
            _logger.warning(
                "Microgrid %s has an unspecified status. Assuming it is active.", self
            )
        return self.status in (MicrogridStatus.ACTIVE, MicrogridStatus.UNSPECIFIED)

    def __str__(self) -> str:
        """Return the ID of this microgrid as a string."""
        name = f":{self.name}" if self.name else ""
        return f"{self.id}{name}"
Attributes¤
create_timestamp instance-attribute ¤
create_timestamp: datetime

The UTC timestamp indicating when the microgrid was initially created.

delivery_area instance-attribute ¤
delivery_area: DeliveryArea | None

The delivery area where the microgrid is located, as identified by a specific code.

enterprise_id instance-attribute ¤
enterprise_id: EnterpriseId

The unique identifier linking this microgrid to its parent enterprise account.

id instance-attribute ¤
id: MicrogridId

The unique identifier of the microgrid.

is_active cached property ¤
is_active: bool

Whether the microgrid is active.

location instance-attribute ¤
location: Location | None

Physical location of the microgrid, in geographical co-ordinates.

name instance-attribute ¤
name: str | None

Name of the microgrid.

status instance-attribute ¤
status: MicrogridStatus | int

The current status of the microgrid.

Functions¤
__str__ ¤
__str__() -> str

Return the ID of this microgrid as a string.

Source code in frequenz/client/microgrid/_microgrid_info.py 
84
85
86
87
def __str__(self) -> str:
    """Return the ID of this microgrid as a string."""
    name = f":{self.name}" if self.name else ""
    return f"{self.id}{name}"

frequenz.client.microgrid.MicrogridStatus ¤

Bases: Enum

The possible statuses for a microgrid.

Source code in frequenz/client/microgrid/_microgrid_info.py 
21
22
23
24
25
26
27
28
29
30
31
32
@enum.unique
class MicrogridStatus(enum.Enum):
    """The possible statuses for a microgrid."""

    UNSPECIFIED = microgrid_pb2.MICROGRID_STATUS_UNSPECIFIED
    """The status is unspecified. This should not be used."""

    ACTIVE = microgrid_pb2.MICROGRID_STATUS_ACTIVE
    """The microgrid is active."""

    INACTIVE = microgrid_pb2.MICROGRID_STATUS_INACTIVE
    """The microgrid is inactive."""
Attributes¤
ACTIVE class-attribute instance-attribute ¤
ACTIVE = MICROGRID_STATUS_ACTIVE

The microgrid is active.

INACTIVE class-attribute instance-attribute ¤
INACTIVE = MICROGRID_STATUS_INACTIVE

The microgrid is inactive.

UNSPECIFIED class-attribute instance-attribute ¤
UNSPECIFIED = MICROGRID_STATUS_UNSPECIFIED

The status is unspecified. This should not be used.

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 
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
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 
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
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 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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 
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
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 
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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 
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
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 
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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 
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
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 
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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 
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
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 
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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 
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
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 
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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 
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
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 
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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 
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
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 
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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 
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
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 
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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 
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
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 
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
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 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@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.Validity ¤

Bases: Enum

The duration for which a given list of bounds will stay in effect.

Source code in frequenz/client/microgrid/_client.py 
639
640
641
642
643
644
645
646
647
648
649
650
651
652
class Validity(enum.Enum):
    """The duration for which a given list of bounds will stay in effect."""

    FIVE_SECONDS = 5
    """The bounds will stay in effect for 5 seconds."""

    ONE_MINUTE = 60
    """The bounds will stay in effect for 1 minute."""

    FIVE_MINUTES = 60 * 5
    """The bounds will stay in effect for 5 minutes."""

    FIFTEEN_MINUTES = 60 * 15
    """The bounds will stay in effect for 15 minutes."""
Attributes¤
FIFTEEN_MINUTES class-attribute instance-attribute ¤
FIFTEEN_MINUTES = 60 * 15

The bounds will stay in effect for 15 minutes.

FIVE_MINUTES class-attribute instance-attribute ¤
FIVE_MINUTES = 60 * 5

The bounds will stay in effect for 5 minutes.

FIVE_SECONDS class-attribute instance-attribute ¤
FIVE_SECONDS = 5

The bounds will stay in effect for 5 seconds.

ONE_MINUTE class-attribute instance-attribute ¤
ONE_MINUTE = 60

The bounds will stay in effect for 1 minute.