Skip to content

microgrid

frequenz.client.microgrid ¤

Client to connect to the Microgrid API.

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

Classes¤

frequenz.client.microgrid.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.BatteryComponentState ¤

Bases: Enum

Component states of a battery.

Source code in frequenz/client/microgrid/_component_states.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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class BatteryComponentState(Enum):
    """Component states of a battery."""

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

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

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

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

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

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

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

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

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

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

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

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

        Args:
            state: The protobuf component state to convert.

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

The battery is consuming electrical energy.

DISCHARGING class-attribute instance-attribute ¤
DISCHARGING = COMPONENT_STATE_DISCHARGING

The battery is generating electrical energy.

ERROR class-attribute instance-attribute ¤
ERROR = COMPONENT_STATE_ERROR

The battery is in a faulty state.

IDLE class-attribute instance-attribute ¤
IDLE = COMPONENT_STATE_IDLE

The battery is idle.

LOCKED class-attribute instance-attribute ¤
LOCKED = COMPONENT_STATE_LOCKED

The battery is online, but currently unavailable.

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

OFF class-attribute instance-attribute ¤
OFF = COMPONENT_STATE_OFF

The battery is switched off.

SWITCHING_OFF class-attribute instance-attribute ¤
SWITCHING_OFF = COMPONENT_STATE_SWITCHING_OFF

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

SWITCHING_ON class-attribute instance-attribute ¤
SWITCHING_ON = COMPONENT_STATE_SWITCHING_ON

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

UNKNOWN class-attribute instance-attribute ¤
UNKNOWN = COMPONENT_STATE_UNKNOWN

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

UNSPECIFIED class-attribute instance-attribute ¤
UNSPECIFIED = COMPONENT_STATE_UNSPECIFIED

Unspecified component state.

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

Convert a protobuf state value to this enum.

PARAMETER DESCRIPTION
state

The protobuf component state to convert.

TYPE: ValueType

RETURNS DESCRIPTION
Self

The enum value corresponding to the protobuf message.
Source code in frequenz/client/microgrid/_component_states.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@classmethod
def from_pb(cls, state: battery_pb2.ComponentState.ValueType) -> Self:
    """Convert a protobuf state value to this enum.

    Args:
        state: The protobuf component state to convert.

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

frequenz.client.microgrid.BatteryData dataclass ¤

Bases: ComponentData

A wrapper class for holding battery data.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

component_id instance-attribute ¤
component_id: int

The ID identifying this component in the microgrid.

component_state instance-attribute ¤
component_state: BatteryComponentState

State of the battery.

errors instance-attribute ¤
errors: list[BatteryError]

List of errors in protobuf struct.

power_exclusion_lower_bound instance-attribute ¤
power_exclusion_lower_bound: float

Lower exclusion bound for battery power in watts.

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

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

power_exclusion_upper_bound instance-attribute ¤
power_exclusion_upper_bound: float

Upper exclusion bound for battery power in watts.

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

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

power_inclusion_lower_bound instance-attribute ¤
power_inclusion_lower_bound: float

Lower inclusion bound for battery power in watts.

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

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

power_inclusion_upper_bound instance-attribute ¤
power_inclusion_upper_bound: float

Upper inclusion bound for battery power in watts.

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

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

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

Raw component data as decoded from the wire.

relay_state instance-attribute ¤
relay_state: BatteryRelayState

State of the battery relay.

soc instance-attribute ¤
soc: float

Battery's overall SoC in percent (%).

soc_lower_bound instance-attribute ¤
soc_lower_bound: float

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

soc_upper_bound instance-attribute ¤
soc_upper_bound: float

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

temperature instance-attribute ¤
temperature: float

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

timestamp instance-attribute ¤
timestamp: datetime

The timestamp of when the data was measured.

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

Create BatteryData from a protobuf message.

PARAMETER DESCRIPTION
raw

raw component data as decoded from the wire.

TYPE: ComponentData

RETURNS DESCRIPTION
Self

Instance of BatteryData created from the protobuf message.
Source code in frequenz/client/microgrid/_component_data.py 
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
@classmethod
def from_proto(cls, raw: microgrid_pb2.ComponentData) -> Self:
    """Create BatteryData from a protobuf message.

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

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

frequenz.client.microgrid.BatteryError dataclass ¤

A battery error.

Source code in frequenz/client/microgrid/_component_error.py 
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
@dataclass(frozen=True, kw_only=True)
class BatteryError:
    """A battery error."""

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

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

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

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

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

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

The error code.

level class-attribute instance-attribute ¤
level: ErrorLevel = UNSPECIFIED

The error level.

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

The error message.

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

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

PARAMETER DESCRIPTION
raw

The protobuf message to get the values from.

TYPE: Error

RETURNS DESCRIPTION
Self

The new instance with the values from the protobuf message.
Source code in frequenz/client/microgrid/_component_error.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
@classmethod
def from_pb(cls, raw: battery_pb2.Error) -> Self:
    """Create a new instance using a protobuf message to get the values.

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

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

frequenz.client.microgrid.BatteryErrorCode ¤

Bases: Enum

Battery error code.

Source code in frequenz/client/microgrid/_component_error.py 
 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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class BatteryErrorCode(Enum):
    """Battery error code."""

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

        Args:
            code: The protobuf error code to convert.

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

One or more battery blocks have failed.

CALIBRATION_NEEDED class-attribute instance-attribute ¤
CALIBRATION_NEEDED = ERROR_CODE_CALIBRATION_NEEDED

The battery requires a calibration to reset its measurements.

CONTROLLER_ERROR class-attribute instance-attribute ¤
CONTROLLER_ERROR = ERROR_CODE_CONTROLLER_ERROR

The battery controller has failed.

EXCEEDED_SOP_CHARGE class-attribute instance-attribute ¤
EXCEEDED_SOP_CHARGE = ERROR_CODE_EXCEEDED_SOP_CHARGE

Charge current has exceeded component bounds.

EXCEEDED_SOP_DISCHARGE class-attribute instance-attribute ¤
EXCEEDED_SOP_DISCHARGE = ERROR_CODE_EXCEEDED_SOP_DISCHARGE

Discharge current has exceeded component bounds.

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

The eternal power switch has failed.

FUSE_ERROR class-attribute instance-attribute ¤
FUSE_ERROR = ERROR_CODE_FUSE_ERROR

The battery's fuse has failed.

HIGH_CURRENT_CHARGE class-attribute instance-attribute ¤
HIGH_CURRENT_CHARGE = ERROR_CODE_HIGH_CURRENT_CHARGE

Charge current is too high.

HIGH_CURRENT_DISCHARGE class-attribute instance-attribute ¤
HIGH_CURRENT_DISCHARGE = ERROR_CODE_HIGH_CURRENT_DISCHARGE

Discharge current is too high.

HIGH_HUMIDITY class-attribute instance-attribute ¤
HIGH_HUMIDITY = ERROR_CODE_HIGH_HUMIDITY

Humidity is too high.

HIGH_TEMPERATURE class-attribute instance-attribute ¤
HIGH_TEMPERATURE = ERROR_CODE_HIGH_TEMPERATURE

Temperature is too high.

HIGH_VOLTAGE class-attribute instance-attribute ¤
HIGH_VOLTAGE = ERROR_CODE_HIGH_VOLTAGE

Voltage is too high.

LOW_SOH class-attribute instance-attribute ¤
LOW_SOH = ERROR_CODE_LOW_SOH

The State of health is low.

LOW_TEMPERATURE class-attribute instance-attribute ¤
LOW_TEMPERATURE = ERROR_CODE_LOW_TEMPERATURE

Temperature is too low.

LOW_VOLTAGE class-attribute instance-attribute ¤
LOW_VOLTAGE = ERROR_CODE_LOW_VOLTAGE

Voltage is too low.

PRECHARGE_ERROR class-attribute instance-attribute ¤
PRECHARGE_ERROR = ERROR_CODE_PRECHARGE_ERROR

The precharge operation has failed.

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

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

RELAY_ERROR class-attribute instance-attribute ¤
RELAY_ERROR = ERROR_CODE_RELAY_ERROR

The battery's DC relays have failed.

SYSTEM_IMBALANCE class-attribute instance-attribute ¤
SYSTEM_IMBALANCE = ERROR_CODE_SYSTEM_IMBALANCE

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

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

System plausibility checks have failed.

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

System shut down due to extremely low voltage.

UNSPECIFIED class-attribute instance-attribute ¤
UNSPECIFIED = ERROR_CODE_UNSPECIFIED

Unspecified battery error code.

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

Convert a protobuf error code value to this enum.

PARAMETER DESCRIPTION
code

The protobuf error code to convert.

TYPE: ValueType

RETURNS DESCRIPTION
Self

The enum value corresponding to the protobuf message.
Source code in frequenz/client/microgrid/_component_error.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@classmethod
def from_pb(cls, code: battery_pb2.ErrorCode.ValueType) -> Self:
    """Convert a protobuf error code value to this enum.

    Args:
        code: The protobuf error code to convert.

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

frequenz.client.microgrid.BatteryRelayState ¤

Bases: Enum

Relay states of a battery.

Source code in frequenz/client/microgrid/_component_states.py 
 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
class BatteryRelayState(Enum):
    """Relay states of a battery."""

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

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

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

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

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

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

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

        Args:
            state: The protobuf component state to convert.

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

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

ERROR class-attribute instance-attribute ¤
ERROR = RELAY_STATE_ERROR

The relays are in an error state.

LOCKED class-attribute instance-attribute ¤
LOCKED = RELAY_STATE_LOCKED

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

OPENED class-attribute instance-attribute ¤
OPENED = RELAY_STATE_OPENED

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

PRECHARGING class-attribute instance-attribute ¤
PRECHARGING = RELAY_STATE_PRECHARGING

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

UNSPECIFIED class-attribute instance-attribute ¤
UNSPECIFIED = RELAY_STATE_UNSPECIFIED

Unspecified relay state.

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

Convert a protobuf state value to this enum.

PARAMETER DESCRIPTION
state

The protobuf component state to convert.

TYPE: ValueType

RETURNS DESCRIPTION
Self

The enum value corresponding to the protobuf message.
Source code in frequenz/client/microgrid/_component_states.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@classmethod
def from_pb(cls, state: battery_pb2.RelayState.ValueType) -> Self:
    """Convert a protobuf state value to this enum.

    Args:
        state: The protobuf component state to convert.

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

frequenz.client.microgrid.ClientNotConnected ¤

Bases: ApiClientError

The client is not connected to the server.

Source code in frequenz/client/base/exception.py 
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.Component dataclass ¤

Metadata for a single microgrid component.

Source code in frequenz/client/microgrid/_component.py 
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
@dataclass(frozen=True)
class Component:
    """Metadata for a single microgrid component."""

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

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

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

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

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

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

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

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

The category of this component.

component_id instance-attribute ¤
component_id: int

The ID of this component.

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

The metadata of this component.

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

The type of this component.

Functions¤
__hash__ ¤
__hash__() -> int

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

RETURNS DESCRIPTION
int

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

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

Check if this instance contains valid data.

RETURNS DESCRIPTION
bool

True if id > 0 and type is a valid ComponentCategory, or if id == 0 and type is GRID, False otherwise
Source code in frequenz/client/microgrid/_component.py 
175
176
177
178
179
180
181
182
183
184
def is_valid(self) -> bool:
    """Check if this instance contains valid data.

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

frequenz.client.microgrid.ComponentCategory ¤

Bases: Enum

Possible types of microgrid component.

Source code in frequenz/client/microgrid/_component.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
class ComponentCategory(Enum):
    """Possible types of microgrid component."""

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

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

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

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

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

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

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

Battery component.

CHP class-attribute instance-attribute ¤
CHP = COMPONENT_CATEGORY_CHP

CHP component.

EV_CHARGER class-attribute instance-attribute ¤
EV_CHARGER = COMPONENT_CATEGORY_EV_CHARGER

EV charger component.

GRID class-attribute instance-attribute ¤
GRID = COMPONENT_CATEGORY_GRID

Grid component.

INVERTER class-attribute instance-attribute ¤
INVERTER = COMPONENT_CATEGORY_INVERTER

Inverter component.

METER class-attribute instance-attribute ¤
METER = COMPONENT_CATEGORY_METER

Meter component.

NONE class-attribute instance-attribute ¤
NONE = COMPONENT_CATEGORY_UNSPECIFIED

Unspecified component category.

frequenz.client.microgrid.ComponentData dataclass ¤

Bases: ABC

A private base class for strongly typed component data classes.

Source code in frequenz/client/microgrid/_component_data.py 
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
@dataclass(frozen=True)
class ComponentData(ABC):
    """A private base class for strongly typed component data classes."""

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

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

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

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

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

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

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

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

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

The ID identifying this component in the microgrid.

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

Raw component data as decoded from the wire.

timestamp instance-attribute ¤
timestamp: datetime

The timestamp of when the data was measured.

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

Create ComponentData from a protobuf message.

PARAMETER DESCRIPTION
raw

raw component data as decoded from the wire.

TYPE: ComponentData

RETURNS DESCRIPTION
Self

The instance created from the protobuf message.
Source code in frequenz/client/microgrid/_component_data.py 
53
54
55
56
57
58
59
60
61
62
63
@classmethod
@abstractmethod
def from_proto(cls, raw: microgrid_pb2.ComponentData) -> Self:
    """Create ComponentData from a protobuf message.

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

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

frequenz.client.microgrid.ComponentMetadata dataclass ¤

Base class for component metadata classes.

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

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

The fuse at the grid connection point.

frequenz.client.microgrid.ComponentMetricId ¤

Bases: Enum

An enum representing the various metrics available in the microgrid.

Source code in frequenz/client/microgrid/_component.py 
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
class ComponentMetricId(Enum):
    """An enum representing the various metrics available in the microgrid."""

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

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

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

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

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

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

    FREQUENCY = "frequency"

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

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

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

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

Active power.

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

Active power exclusion lower bound.

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

Active power exclusion upper bound.

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

Active power inclusion lower bound.

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

Active power inclusion upper bound.

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

Active power in phase 1.

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

Active power in phase 2.

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

Active power in phase 3.

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

Capacity.

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

Current in phase 1.

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

Current in phase 2.

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

Current in phase 3.

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

Power exclusion lower bound.

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

Power exclusion upper bound.

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

Power inclusion lower bound.

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

Power inclusion upper bound.

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

Reactive power.

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

Reactive power in phase 1.

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

Reactive power in phase 2.

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

Reactive power in phase 3.

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

State of charge.

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

Lower bound of state of charge.

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

Upper bound of state of charge.

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

Temperature.

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

Voltage in phase 1.

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

Voltage in phase 2.

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

Voltage in phase 3.

frequenz.client.microgrid.ComponentType ¤

Bases: Enum

A base class from which individual component types are derived.

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

frequenz.client.microgrid.Connection dataclass ¤

Metadata for a connection between microgrid components.

Source code in frequenz/client/microgrid/_connection.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
@dataclass(frozen=True)
class Connection:
    """Metadata for a connection between microgrid components."""

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

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

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

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

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

start instance-attribute ¤
start: int

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

Functions¤
is_valid ¤
is_valid() -> bool

Check if this instance contains valid data.

RETURNS DESCRIPTION
bool

True if start >= 0, end > 0, and start != end, False otherwise.
Source code in frequenz/client/microgrid/_connection.py 
20
21
22
23
24
25
26
27
def is_valid(self) -> bool:
    """Check if this instance contains valid data.

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

frequenz.client.microgrid.DataLoss ¤

Bases: GrpcError

Unrecoverable data loss or corruption.

Source code in frequenz/client/base/exception.py 
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.EVChargerCableState ¤

Bases: Enum

Cable states of an EV Charger.

Source code in frequenz/client/microgrid/_component_states.py 
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
class EVChargerCableState(Enum):
    """Cable states of an EV Charger."""

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

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

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

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

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

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

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

        Args:
            state: The protobuf cable state to convert.

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

The cable is plugged into the charging station and locked.

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

The cable is plugged into the charging station.

EV_LOCKED class-attribute instance-attribute ¤
EV_LOCKED = CABLE_STATE_EV_LOCKED

The cable is plugged into the EV and locked.

EV_PLUGGED class-attribute instance-attribute ¤
EV_PLUGGED = CABLE_STATE_EV_PLUGGED

The cable is plugged into the EV.

UNPLUGGED class-attribute instance-attribute ¤
UNPLUGGED = CABLE_STATE_UNPLUGGED

The cable is unplugged.

UNSPECIFIED class-attribute instance-attribute ¤
UNSPECIFIED = CABLE_STATE_UNSPECIFIED

Unspecified cable state.

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

Convert a protobuf state value to this enum.

PARAMETER DESCRIPTION
state

The protobuf cable state to convert.

TYPE: ValueType

RETURNS DESCRIPTION
Self

The enum value corresponding to the protobuf message.
Source code in frequenz/client/microgrid/_component_states.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
@classmethod
def from_pb(cls, state: ev_charger_pb2.CableState.ValueType) -> Self:
    """Convert a protobuf state value to this enum.

    Args:
        state: The protobuf cable state to convert.

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

frequenz.client.microgrid.EVChargerComponentState ¤

Bases: Enum

Component State of an EV Charger.

Source code in frequenz/client/microgrid/_component_states.py 
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
class EVChargerComponentState(Enum):
    """Component State of an EV Charger."""

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

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

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

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

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

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

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

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

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

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

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

        Args:
            state: The protobuf component state to convert.

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

The component rejected authorization.

CHARGING class-attribute instance-attribute ¤
CHARGING = COMPONENT_STATE_CHARGING

The component is charging.

DISCHARGING class-attribute instance-attribute ¤
DISCHARGING = COMPONENT_STATE_DISCHARGING

The component is discharging.

ERROR class-attribute instance-attribute ¤
ERROR = COMPONENT_STATE_ERROR

The component is in error state.

INTERRUPTED class-attribute instance-attribute ¤
INTERRUPTED = COMPONENT_STATE_INTERRUPTED

The component is interrupted.

NOT_READY class-attribute instance-attribute ¤
NOT_READY = COMPONENT_STATE_NOT_READY

The component is not ready.

READY class-attribute instance-attribute ¤
READY = COMPONENT_STATE_READY

The component is ready.

STARTING class-attribute instance-attribute ¤
STARTING = COMPONENT_STATE_STARTING

The component is starting.

UNKNOWN class-attribute instance-attribute ¤
UNKNOWN = COMPONENT_STATE_UNKNOWN

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

UNSPECIFIED class-attribute instance-attribute ¤
UNSPECIFIED = COMPONENT_STATE_UNSPECIFIED

Unspecified component state.

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

Convert a protobuf state value to this enum.

PARAMETER DESCRIPTION
state

The protobuf component state to convert.

TYPE: ValueType

RETURNS DESCRIPTION
Self

The enum value corresponding to the protobuf message.
Source code in frequenz/client/microgrid/_component_states.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
@classmethod
def from_pb(cls, state: ev_charger_pb2.ComponentState.ValueType) -> Self:
    """Convert a protobuf state value to this enum.

    Args:
        state: The protobuf component state to convert.

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

frequenz.client.microgrid.EVChargerData dataclass ¤

Bases: ComponentData

A wrapper class for holding ev_charger data.

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

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

    Represented in the passive sign convention.

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

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

    Represented in the passive sign convention.

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

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

    Represented in the passive sign convention.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Represented in the passive sign convention.

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

Lower exclusion bound for EV charger power in watts.

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

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

active_power_exclusion_upper_bound instance-attribute ¤
active_power_exclusion_upper_bound: float

Upper exclusion bound for EV charger power in watts.

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

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

active_power_inclusion_lower_bound instance-attribute ¤
active_power_inclusion_lower_bound: float

Lower inclusion bound for EV charger power in watts.

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

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

active_power_inclusion_upper_bound instance-attribute ¤
active_power_inclusion_upper_bound: float

Upper inclusion bound for EV charger power in watts.

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

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

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

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

Represented in the passive sign convention.

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

The state of the ev charger's cable.

component_id instance-attribute ¤
component_id: int

The ID identifying this component in the microgrid.

component_state instance-attribute ¤
component_state: EVChargerComponentState

The state of the ev charger.

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

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

Represented in the passive sign convention.

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

AC frequency, in Hertz (Hz).

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

Raw component data as decoded from the wire.

reactive_power instance-attribute ¤
reactive_power: float

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

  • Positive power means capacitive (current leading w.r.t. voltage).
  • Negative power means inductive (current lagging w.r.t. voltage).
reactive_power_per_phase instance-attribute ¤
reactive_power_per_phase: tuple[float, float, float]

The per-phase AC reactive power, in Volt-Ampere Reactive (VAr).

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

  • Positive power means capacitive (current leading w.r.t. voltage).
  • Negative power means inductive (current lagging w.r.t. voltage).
timestamp instance-attribute ¤
timestamp: datetime

The timestamp of when the data was measured.

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

The AC voltage in Volts (V) between the line and the neutral wire for phase/line 1,2 and 3 respectively.

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

Create EVChargerData from a protobuf message.

PARAMETER DESCRIPTION
raw

raw component data as decoded from the wire.

TYPE: ComponentData

RETURNS DESCRIPTION
Self

Instance of EVChargerData created from the protobuf message.
Source code in frequenz/client/microgrid/_component_data.py 
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
@classmethod
def from_proto(cls, raw: microgrid_pb2.ComponentData) -> Self:
    """Create EVChargerData from a protobuf message.

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

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

Check whether an EV is connected to the charger.

RETURNS DESCRIPTION
bool

When the charger is not in an error state, whether an EV is connected to the charger.
Source code in frequenz/client/microgrid/_component_data.py 
581
582
583
584
585
586
587
588
589
590
591
592
593
594
def is_ev_connected(self) -> bool:
    """Check whether an EV is connected to the charger.

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

frequenz.client.microgrid.EntityAlreadyExists ¤

Bases: GrpcError

The entity that we attempted to create already exists.

Source code in frequenz/client/base/exception.py 
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.ErrorLevel ¤

Bases: Enum

Error level.

Source code in frequenz/client/microgrid/_component_error.py 
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
class ErrorLevel(Enum):
    """Error level."""

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

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

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

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

        Args:
            code: The protobuf error level to convert.

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

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

UNSPECIFIED class-attribute instance-attribute ¤
UNSPECIFIED = ERROR_LEVEL_UNSPECIFIED

Unspecified component error.

WARN class-attribute instance-attribute ¤
WARN = ERROR_LEVEL_WARN

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

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

Convert a protobuf error level value to this enum.

PARAMETER DESCRIPTION
code

The protobuf error level to convert.

TYPE: ValueType

RETURNS DESCRIPTION
Self

The enum value corresponding to the protobuf message.
Source code in frequenz/client/microgrid/_component_error.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
@classmethod
def from_pb(cls, code: common_pb2.ErrorLevel.ValueType) -> Self:
    """Convert a protobuf error level value to this enum.

    Args:
        code: The protobuf error level to convert.

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

frequenz.client.microgrid.Fuse dataclass ¤

Fuse data class.

Source code in frequenz/client/microgrid/_component.py 
115
116
117
118
119
120
@dataclass(frozen=True)
class Fuse:
    """Fuse data class."""

    max_current: float
    """Rated current of the fuse."""
Attributes¤
max_current instance-attribute ¤
max_current: float

Rated current of the fuse.

frequenz.client.microgrid.GridMetadata dataclass ¤

Bases: ComponentMetadata

Metadata for a grid connection point.

Source code in frequenz/client/microgrid/_component.py 
131
132
133
@dataclass(frozen=True)
class GridMetadata(ComponentMetadata):
    """Metadata for a grid connection point."""
Attributes¤
fuse class-attribute instance-attribute ¤
fuse: Fuse | None = None

The fuse at the grid connection point.

frequenz.client.microgrid.GrpcError ¤

Bases: ApiClientError

The gRPC server returned an error with a status code.

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

The following sub-classes are available:

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

Bases: Enum

Component states of an inverter.

Source code in frequenz/client/microgrid/_component_states.py 
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
class InverterComponentState(Enum):
    """Component states of an inverter."""

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

        Args:
            state: The protobuf component state to convert.

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

The inverter is consuming electrical energy to charge batteries.

Applicable to BATTERY and HYBRID inverters only.

DISCHARGING class-attribute instance-attribute ¤
DISCHARGING = COMPONENT_STATE_DISCHARGING

The inverter is generating electrical energy.

ERROR class-attribute instance-attribute ¤
ERROR = COMPONENT_STATE_ERROR

The inverter is in a faulty state.

IDLE class-attribute instance-attribute ¤
IDLE = COMPONENT_STATE_IDLE

The inverter is idle.

OFF class-attribute instance-attribute ¤
OFF = COMPONENT_STATE_OFF

Inverter is switched off.

STANDBY class-attribute instance-attribute ¤
STANDBY = COMPONENT_STATE_STANDBY

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

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

SWITCHING_OFF class-attribute instance-attribute ¤
SWITCHING_OFF = COMPONENT_STATE_SWITCHING_OFF

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

SWITCHING_ON class-attribute instance-attribute ¤
SWITCHING_ON = COMPONENT_STATE_SWITCHING_ON

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

UNAVAILABLE class-attribute instance-attribute ¤
UNAVAILABLE = COMPONENT_STATE_UNAVAILABLE

The inverter is online, but currently unavailable.

Possibly due to a pre- scheduled maintenance.

UNKNOWN class-attribute instance-attribute ¤
UNKNOWN = COMPONENT_STATE_UNKNOWN

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

UNSPECIFIED class-attribute instance-attribute ¤
UNSPECIFIED = COMPONENT_STATE_UNSPECIFIED

Unspecified component state.

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

Convert a protobuf state value to this enum.

PARAMETER DESCRIPTION
state

The protobuf component state to convert.

TYPE: ValueType

RETURNS DESCRIPTION
Self

The enum value corresponding to the protobuf message.
Source code in frequenz/client/microgrid/_component_states.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
@classmethod
def from_pb(cls, state: inverter_pb2.ComponentState.ValueType) -> Self:
    """Convert a protobuf state value to this enum.

    Args:
        state: The protobuf component state to convert.

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

frequenz.client.microgrid.InverterData dataclass ¤

Bases: ComponentData

A wrapper class for holding inverter data.

Source code in frequenz/client/microgrid/_component_data.py 
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
@dataclass(frozen=True)
class InverterData(ComponentData):  # pylint: disable=too-many-instance-attributes
    """A wrapper class for holding inverter data."""

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

    Represented in the passive sign convention.

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

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

    Represented in the passive sign convention.

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

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

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

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

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

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

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

    Represented in the passive sign convention.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

        Returns:
            Instance of InverterData created from the protobuf message.
        """
        raw_power = raw.inverter.data.ac.power_active
        inverter_data = cls(
            component_id=raw.id,
            timestamp=raw.ts.ToDatetime(tzinfo=timezone.utc),
            active_power=raw.inverter.data.ac.power_active.value,
            active_power_per_phase=(
                raw.inverter.data.ac.phase_1.power_active.value,
                raw.inverter.data.ac.phase_2.power_active.value,
                raw.inverter.data.ac.phase_3.power_active.value,
            ),
            reactive_power=raw.inverter.data.ac.power_reactive.value,
            reactive_power_per_phase=(
                raw.inverter.data.ac.phase_1.power_reactive.value,
                raw.inverter.data.ac.phase_2.power_reactive.value,
                raw.inverter.data.ac.phase_3.power_reactive.value,
            ),
            current_per_phase=(
                raw.inverter.data.ac.phase_1.current.value,
                raw.inverter.data.ac.phase_2.current.value,
                raw.inverter.data.ac.phase_3.current.value,
            ),
            voltage_per_phase=(
                raw.inverter.data.ac.phase_1.voltage.value,
                raw.inverter.data.ac.phase_2.voltage.value,
                raw.inverter.data.ac.phase_3.voltage.value,
            ),
            active_power_inclusion_lower_bound=raw_power.system_inclusion_bounds.lower,
            active_power_exclusion_lower_bound=raw_power.system_exclusion_bounds.lower,
            active_power_inclusion_upper_bound=raw_power.system_inclusion_bounds.upper,
            active_power_exclusion_upper_bound=raw_power.system_exclusion_bounds.upper,
            frequency=raw.inverter.data.ac.frequency.value,
            component_state=InverterComponentState.from_pb(
                raw.inverter.state.component_state
            ),
            errors=[InverterError.from_pb(e) for e in raw.inverter.errors],
        )

        inverter_data._set_raw(raw=raw)
        return inverter_data
Attributes¤
active_power instance-attribute ¤
active_power: float

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

Represented in the passive sign convention.

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

Lower exclusion bound for inverter power in watts.

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

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

active_power_exclusion_upper_bound instance-attribute ¤
active_power_exclusion_upper_bound: float

Upper exclusion bound for inverter power in watts.

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

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

active_power_inclusion_lower_bound instance-attribute ¤
active_power_inclusion_lower_bound: float

Lower inclusion bound for inverter power in watts.

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

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

active_power_inclusion_upper_bound instance-attribute ¤
active_power_inclusion_upper_bound: float

Upper inclusion bound for inverter power in watts.

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

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

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

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

Represented in the passive sign convention.

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

The ID identifying this component in the microgrid.

component_state instance-attribute ¤
component_state: InverterComponentState

State of the inverter.

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

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

Represented in the passive sign convention.

  • Positive means consumption from the grid.
  • Negative means supply into the grid.
errors instance-attribute ¤
errors: list[InverterError]

List of errors from the component.

frequency instance-attribute ¤
frequency: float

AC frequency, in Hertz (Hz).

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

Raw component data as decoded from the wire.

reactive_power instance-attribute ¤
reactive_power: float

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

  • Positive power means capacitive (current leading w.r.t. voltage).
  • Negative power means inductive (current lagging w.r.t. voltage).
reactive_power_per_phase instance-attribute ¤
reactive_power_per_phase: tuple[float, float, float]

The per-phase AC reactive power, in Volt-Ampere Reactive (VAr).

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

  • Positive power means capacitive (current leading w.r.t. voltage).
  • Negative power means inductive (current lagging w.r.t. voltage).
timestamp instance-attribute ¤
timestamp: datetime

The timestamp of when the data was measured.

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

The AC voltage in Volts (V) between the line and the neutral wire for phase/line 1, 2 and 3 respectively.

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

Create InverterData from a protobuf message.

PARAMETER DESCRIPTION
raw

raw component data as decoded from the wire.

TYPE: ComponentData

RETURNS DESCRIPTION
Self

Instance of InverterData created from the protobuf message.
Source code in frequenz/client/microgrid/_component_data.py 
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
@classmethod
def from_proto(cls, raw: microgrid_pb2.ComponentData) -> Self:
    """Create InverterData from a protobuf message.

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

    Returns:
        Instance of InverterData created from the protobuf message.
    """
    raw_power = raw.inverter.data.ac.power_active
    inverter_data = cls(
        component_id=raw.id,
        timestamp=raw.ts.ToDatetime(tzinfo=timezone.utc),
        active_power=raw.inverter.data.ac.power_active.value,
        active_power_per_phase=(
            raw.inverter.data.ac.phase_1.power_active.value,
            raw.inverter.data.ac.phase_2.power_active.value,
            raw.inverter.data.ac.phase_3.power_active.value,
        ),
        reactive_power=raw.inverter.data.ac.power_reactive.value,
        reactive_power_per_phase=(
            raw.inverter.data.ac.phase_1.power_reactive.value,
            raw.inverter.data.ac.phase_2.power_reactive.value,
            raw.inverter.data.ac.phase_3.power_reactive.value,
        ),
        current_per_phase=(
            raw.inverter.data.ac.phase_1.current.value,
            raw.inverter.data.ac.phase_2.current.value,
            raw.inverter.data.ac.phase_3.current.value,
        ),
        voltage_per_phase=(
            raw.inverter.data.ac.phase_1.voltage.value,
            raw.inverter.data.ac.phase_2.voltage.value,
            raw.inverter.data.ac.phase_3.voltage.value,
        ),
        active_power_inclusion_lower_bound=raw_power.system_inclusion_bounds.lower,
        active_power_exclusion_lower_bound=raw_power.system_exclusion_bounds.lower,
        active_power_inclusion_upper_bound=raw_power.system_inclusion_bounds.upper,
        active_power_exclusion_upper_bound=raw_power.system_exclusion_bounds.upper,
        frequency=raw.inverter.data.ac.frequency.value,
        component_state=InverterComponentState.from_pb(
            raw.inverter.state.component_state
        ),
        errors=[InverterError.from_pb(e) for e in raw.inverter.errors],
    )

    inverter_data._set_raw(raw=raw)
    return inverter_data

frequenz.client.microgrid.InverterError dataclass ¤

An inverter error.

Source code in frequenz/client/microgrid/_component_error.py 
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
@dataclass(frozen=True, kw_only=True)
class InverterError:
    """An inverter error."""

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

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

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

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

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

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

The error code.

level class-attribute instance-attribute ¤
level: ErrorLevel = UNSPECIFIED

The error level.

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

The error message.

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

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

PARAMETER DESCRIPTION
raw

The protobuf message to get the values from.

TYPE: Error

RETURNS DESCRIPTION
Self

The new instance with the values from the protobuf message.
Source code in frequenz/client/microgrid/_component_error.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
@classmethod
def from_pb(cls, raw: inverter_pb2.Error) -> Self:
    """Create a new instance using a protobuf message to get the values.

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

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

frequenz.client.microgrid.InverterErrorCode ¤

Bases: Enum

Inverter error code.

Source code in frequenz/client/microgrid/_component_error.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
class InverterErrorCode(Enum):
    """Inverter error code."""

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

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

        Args:
            code: The protobuf error code to convert.

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

Unspecified inverter error code.

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

Convert a protobuf error code value to this enum.

PARAMETER DESCRIPTION
code

The protobuf error code to convert.

TYPE: ValueType

RETURNS DESCRIPTION
Self

The enum value corresponding to the protobuf message.
Source code in frequenz/client/microgrid/_component_error.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
@classmethod
def from_pb(cls, code: inverter_pb2.ErrorCode.ValueType) -> Self:
    """Convert a protobuf error code value to this enum.

    Args:
        code: The protobuf error code to convert.

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

frequenz.client.microgrid.InverterType ¤

Bases: ComponentType

Enum representing inverter types.

Source code in frequenz/client/microgrid/_component.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class InverterType(ComponentType):
    """Enum representing inverter types."""

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

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

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

    HYBRID = inverter_pb2.Type.TYPE_HYBRID
    """Hybrid inverter."""
Attributes¤
BATTERY class-attribute instance-attribute ¤
BATTERY = TYPE_BATTERY

Battery inverter.

HYBRID class-attribute instance-attribute ¤
HYBRID = TYPE_HYBRID

Hybrid inverter.

NONE class-attribute instance-attribute ¤
NONE = TYPE_UNSPECIFIED

Unspecified inverter type.

SOLAR class-attribute instance-attribute ¤
SOLAR = TYPE_SOLAR

Solar inverter.

frequenz.client.microgrid.Location dataclass ¤

Metadata for the location of microgrid.

Source code in frequenz/client/microgrid/_metadata.py 
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
@dataclass(frozen=True, kw_only=True)
class Location:
    """Metadata for the location of microgrid."""

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

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

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

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

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

        timezone = _timezone_finder.timezone_at(lat=self.latitude, lng=self.longitude)
        if timezone:
            # The dataclass is frozen, so it needs to use __setattr__ to set the timezone.
            object.__setattr__(self, "timezone", ZoneInfo(key=timezone))
Attributes¤
latitude class-attribute instance-attribute ¤
latitude: float | None = None

The latitude of the microgrid in degree.

longitude class-attribute instance-attribute ¤
longitude: float | None = None

The longitude of the microgrid in degree.

timezone class-attribute instance-attribute ¤
timezone: ZoneInfo | None = None

The timezone of the microgrid.

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

Functions¤
__post_init__ ¤
__post_init__() -> None

Initialize the timezone of the microgrid.

Source code in frequenz/client/microgrid/_metadata.py 
33
34
35
36
37
38
39
40
41
def __post_init__(self) -> None:
    """Initialize the timezone of the microgrid."""
    if self.latitude is None or self.longitude is None or self.timezone is not None:
        return

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

frequenz.client.microgrid.Metadata dataclass ¤

Metadata for the microgrid.

Source code in frequenz/client/microgrid/_metadata.py 
44
45
46
47
48
49
50
51
52
@dataclass(frozen=True, kw_only=True)
class Metadata:
    """Metadata for the microgrid."""

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

    location: Location | None = None
    """The location of the microgrid."""
Attributes¤
location class-attribute instance-attribute ¤
location: Location | None = None

The location of the microgrid.

microgrid_id class-attribute instance-attribute ¤
microgrid_id: int | None = None

The ID of the microgrid.

frequenz.client.microgrid.MeterData dataclass ¤

Bases: ComponentData

A wrapper class for holding meter data.

Source code in frequenz/client/microgrid/_component_data.py 
 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
@dataclass(frozen=True)
class MeterData(ComponentData):
    """A wrapper class for holding meter data."""

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

    Represented in the passive sign convention.

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

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

    Represented in the passive sign convention.

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

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

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

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

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

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

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

    Represented in the passive sign convention.

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

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

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

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

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

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

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

Represented in the passive sign convention.

  • Positive means consumption from the grid.
  • Negative means supply into the grid.
active_power_per_phase instance-attribute ¤
active_power_per_phase: tuple[float, float, float]

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

Represented in the passive sign convention.

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

The ID identifying this component in the microgrid.

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

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

Represented in the passive sign convention.

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

The AC power frequency in Hertz (Hz).

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

Raw component data as decoded from the wire.

reactive_power instance-attribute ¤
reactive_power: float

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

  • Positive power means capacitive (current leading w.r.t. voltage).
  • Negative power means inductive (current lagging w.r.t. voltage).
reactive_power_per_phase instance-attribute ¤
reactive_power_per_phase: tuple[float, float, float]

The per-phase AC reactive power, in Volt-Ampere Reactive (VAr).

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

  • Positive power means capacitive (current leading w.r.t. voltage).
  • Negative power means inductive (current lagging w.r.t. voltage).
timestamp instance-attribute ¤
timestamp: datetime

The timestamp of when the data was measured.

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

The ac voltage in volts (v) between the line and the neutral wire for phase/line 1,2 and 3 respectively.

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

Create MeterData from a protobuf message.

PARAMETER DESCRIPTION
raw

raw component data as decoded from the wire.

TYPE: ComponentData

RETURNS DESCRIPTION
Self

Instance of MeterData created from the protobuf message.
Source code in frequenz/client/microgrid/_component_data.py 
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
@classmethod
def from_proto(cls, raw: microgrid_pb2.ComponentData) -> Self:
    """Create MeterData from a protobuf message.

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

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

frequenz.client.microgrid.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
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._broadcasters: dict[int, 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

    async def components(  # noqa: DOC502 (raises ApiClientError indirectly)
        self,
    ) -> Iterable[Component]:
        """Fetch all the components present in the microgrid.

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

        Raises:
            ApiClientError: If the are any errors communicating with the Microgrid API,
                most likely a subclass of
                [GrpcError][frequenz.client.microgrid.GrpcError].
        """
        component_list = await client.call_stub_method(
            self,
            lambda: self.stub.ListComponents(
                microgrid_pb2.ComponentFilter(),
                timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
            ),
            method_name="ListComponents",
        )

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

        return result

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

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

        Returns:
            the microgrid metadata.
        """
        microgrid_metadata: microgrid_pb2.MicrogridMetadata | None = None
        try:
            microgrid_metadata = await client.call_stub_method(
                self,
                lambda: self.stub.GetMicrogridMetadata(
                    Empty(),
                    timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
                ),
                method_name="GetMicrogridMetadata",
            )
        except ApiClientError:
            _logger.exception("The microgrid metadata is not available.")

        if not microgrid_metadata:
            return Metadata()

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

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

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

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

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

        Raises:
            ApiClientError: If the are any errors communicating with the Microgrid API,
                most likely a subclass of
                [GrpcError][frequenz.client.microgrid.GrpcError].
        """
        connection_filter = microgrid_pb2.ConnectionFilter(starts=starts, ends=ends)
        valid_components, all_connections = await asyncio.gather(
            self.components(),
            client.call_stub_method(
                self,
                lambda: self.stub.ListConnections(
                    connection_filter,
                    timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
                ),
                method_name="ListConnections",
            ),
        )

        # Filter out the components filtered in `components` method.
        # id=0 is an exception indicating grid component.
        valid_ids = {c.component_id for c in valid_components}
        valid_ids.add(0)

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

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

        return result

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

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

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

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

        broadcaster = self._broadcasters.get(component_id)
        if broadcaster is None:
            broadcaster = streaming.GrpcStreamBroadcaster(
                f"raw-component-data-{component_id}",
                lambda: aiter(
                    self.stub.StreamComponentData(
                        microgrid_pb2.ComponentIdParam(id=component_id)
                    )
                ),
                transform,
                retry_strategy=self._retry_strategy,
                # We don't expect any data stream to end, so if it is exhausted for any
                # reason we want to keep retrying
                retry_on_exhausted_stream=True,
            )
            self._broadcasters[component_id] = broadcaster
        return broadcaster.new_receiver(maxsize=maxsize)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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


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

        Raises:
            ApiClientError: If the are any errors communicating with the Microgrid API,
                most likely a subclass of
                [GrpcError][frequenz.client.microgrid.GrpcError].
        """
        await client.call_stub_method(
            self,
            lambda: self.stub.SetPowerActive(
                microgrid_pb2.SetPowerActiveParam(
                    component_id=component_id, power=power_w
                ),
                timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
            ),
            method_name="SetPowerActive",
        )

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

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

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

        Raises:
            ApiClientError: If the are any errors communicating with the Microgrid API,
                most likely a subclass of
                [GrpcError][frequenz.client.microgrid.GrpcError].
        """
        await client.call_stub_method(
            self,
            lambda: self.stub.SetPowerReactive(
                microgrid_pb2.SetPowerReactiveParam(
                    component_id=component_id, power=reactive_power_var
                ),
                timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
            ),
            method_name="SetPowerReactive",
        )

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

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

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

        target_metric = (
            microgrid_pb2.SetBoundsParam.TargetMetric.TARGET_METRIC_POWER_ACTIVE
        )
        await client.call_stub_method(
            self,
            lambda: self.stub.AddInclusionBounds(
                microgrid_pb2.SetBoundsParam(
                    component_id=component_id,
                    target_metric=target_metric,
                    bounds=metrics_pb2.Bounds(lower=lower, upper=upper),
                ),
                timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
            ),
            method_name="AddInclusionBounds",
        )
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 
237
238
239
240
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

Exit a context manager.

Source code in frequenz/client/base/client.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
async def __aexit__(
    self,
    _exc_type: type[BaseException] | None,
    _exc_val: BaseException | None,
    _exc_tb: Any | None,
) -> bool | None:
    """Exit a context manager."""
    if self._channel is None:
        return None
    result = await self._channel.__aexit__(_exc_type, _exc_val, _exc_tb)
    self._channel = None
    self._stub = None
    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
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._broadcasters: dict[int, streaming.GrpcStreamBroadcaster[Any, Any]] = {}
    self._retry_strategy = retry_strategy
battery_data async ¤
battery_data(
    component_id: int, maxsize: int = RECEIVER_MAX_SIZE
) -> Receiver[BatteryData]

Return a channel receiver that provides a BatteryData stream.

RAISES DESCRIPTION
ValueError

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

id of the battery to get data for.

TYPE: int

maxsize

Size of the receiver's buffer.

TYPE: int DEFAULT: RECEIVER_MAX_SIZE

RETURNS DESCRIPTION
Receiver[BatteryData]

A channel receiver that provides realtime battery data.
Source code in frequenz/client/microgrid/_client.py 
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
async def battery_data(  # noqa: DOC502 (ValueError is raised indirectly by _expect_category)
    self,
    component_id: int,
    maxsize: int = RECEIVER_MAX_SIZE,
) -> Receiver[BatteryData]:
    """Return a channel receiver that provides a `BatteryData` stream.

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

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

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

Fetch all the components present in the microgrid.

RETURNS DESCRIPTION
Iterable[Component]

Iterator whose elements are all the components in the microgrid.
RAISES DESCRIPTION
ApiClientError

If the are any errors communicating with the Microgrid API, most likely a subclass of GrpcError.
Source code in frequenz/client/microgrid/_client.py 
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
async def components(  # noqa: DOC502 (raises ApiClientError indirectly)
    self,
) -> Iterable[Component]:
    """Fetch all the components present in the microgrid.

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

    Raises:
        ApiClientError: If the are any errors communicating with the Microgrid API,
            most likely a subclass of
            [GrpcError][frequenz.client.microgrid.GrpcError].
    """
    component_list = await client.call_stub_method(
        self,
        lambda: self.stub.ListComponents(
            microgrid_pb2.ComponentFilter(),
            timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
        ),
        method_name="ListComponents",
    )

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

    return result
connect ¤
connect(server_url: str | None = None) -> 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

Source code in frequenz/client/base/client.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def connect(self, server_url: str | None = None) -> 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.
    """
    if server_url is not None and server_url != self._server_url:  # URL changed
        self._server_url = server_url
    elif self.is_connected:
        return
    self._channel = parse_grpc_uri(self._server_url, self._channel_defaults)
    self._stub = self._create_stub(self._channel)
connections async ¤
connections(
    starts: Set[int] = frozenset(),
    ends: Set[int] = frozenset(),
) -> Iterable[Connection]

Fetch the connections between components in the microgrid.

PARAMETER DESCRIPTION
starts

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

TYPE: Set[int] DEFAULT: frozenset()

ends

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

TYPE: Set[int] DEFAULT: frozenset()

RETURNS DESCRIPTION
Iterable[Connection]

Microgrid connections matching the provided start and end filters.
RAISES DESCRIPTION
ApiClientError

If the are any errors communicating with the Microgrid API, most likely a subclass of GrpcError.
Source code in frequenz/client/microgrid/_client.py 
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
async def connections(  # noqa: DOC502 (raises ApiClientError indirectly)
    self,
    starts: Set[int] = frozenset(),
    ends: Set[int] = frozenset(),
) -> Iterable[Connection]:
    """Fetch the connections between components in the microgrid.

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

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

    Raises:
        ApiClientError: If the are any errors communicating with the Microgrid API,
            most likely a subclass of
            [GrpcError][frequenz.client.microgrid.GrpcError].
    """
    connection_filter = microgrid_pb2.ConnectionFilter(starts=starts, ends=ends)
    valid_components, all_connections = await asyncio.gather(
        self.components(),
        client.call_stub_method(
            self,
            lambda: self.stub.ListConnections(
                connection_filter,
                timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
            ),
            method_name="ListConnections",
        ),
    )

    # Filter out the components filtered in `components` method.
    # id=0 is an exception indicating grid component.
    valid_ids = {c.component_id for c in valid_components}
    valid_ids.add(0)

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

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

    return result
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 
230
231
232
233
234
235
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)
ev_charger_data async ¤
ev_charger_data(
    component_id: int, maxsize: int = RECEIVER_MAX_SIZE
) -> Receiver[EVChargerData]

Return a channel receiver that provides an EvChargeData stream.

RAISES DESCRIPTION
ValueError

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

id of the ev charger to get data for.

TYPE: int

maxsize

Size of the receiver's buffer.

TYPE: int DEFAULT: RECEIVER_MAX_SIZE

RETURNS DESCRIPTION
Receiver[EVChargerData]

A channel receiver that provides realtime ev charger data.
Source code in frequenz/client/microgrid/_client.py 
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
async def ev_charger_data(  # noqa: DOC502 (ValueError is raised indirectly by _expect_category)
    self,
    component_id: int,
    maxsize: int = RECEIVER_MAX_SIZE,
) -> Receiver[EVChargerData]:
    """Return a channel receiver that provides an `EvChargeData` stream.

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

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

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

Return a channel receiver that provides an InverterData stream.

RAISES DESCRIPTION
ValueError

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

id of the inverter to get data for.

TYPE: int

maxsize

Size of the receiver's buffer.

TYPE: int DEFAULT: RECEIVER_MAX_SIZE

RETURNS DESCRIPTION
Receiver[InverterData]

A channel receiver that provides realtime inverter data.
Source code in frequenz/client/microgrid/_client.py 
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
async def inverter_data(  # noqa: DOC502 (ValueError is raised indirectly by _expect_category)
    self,
    component_id: int,
    maxsize: int = RECEIVER_MAX_SIZE,
) -> Receiver[InverterData]:
    """Return a channel receiver that provides an `InverterData` stream.

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

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

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

Fetch the microgrid metadata.

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

RETURNS DESCRIPTION
Metadata

the microgrid metadata.
Source code in frequenz/client/microgrid/_client.py 
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
async def metadata(self) -> Metadata:
    """Fetch the microgrid metadata.

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

    Returns:
        the microgrid metadata.
    """
    microgrid_metadata: microgrid_pb2.MicrogridMetadata | None = None
    try:
        microgrid_metadata = await client.call_stub_method(
            self,
            lambda: self.stub.GetMicrogridMetadata(
                Empty(),
                timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
            ),
            method_name="GetMicrogridMetadata",
        )
    except ApiClientError:
        _logger.exception("The microgrid metadata is not available.")

    if not microgrid_metadata:
        return Metadata()

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

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

Return a channel receiver that provides a MeterData stream.

RAISES DESCRIPTION
ValueError

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

id of the meter to get data for.

TYPE: int

maxsize

Size of the receiver's buffer.

TYPE: int DEFAULT: RECEIVER_MAX_SIZE

RETURNS DESCRIPTION
Receiver[MeterData]

A channel receiver that provides realtime meter data.
Source code in frequenz/client/microgrid/_client.py 
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
async def meter_data(  # noqa: DOC502 (ValueError is raised indirectly by _expect_category)
    self,
    component_id: int,
    maxsize: int = RECEIVER_MAX_SIZE,
) -> Receiver[MeterData]:
    """Return a channel receiver that provides a `MeterData` stream.

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

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

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

Send SetBoundsParams received from a channel to the Microgrid service.

PARAMETER DESCRIPTION
component_id

ID of the component to set bounds for.

TYPE: int

lower

Lower bound to be set for the component.

TYPE: float

upper

Upper bound to be set for the component.

TYPE: float

RAISES DESCRIPTION
ValueError

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

If the are any errors communicating with the Microgrid API, most likely a subclass of GrpcError.
Source code in frequenz/client/microgrid/_client.py 
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
async def set_bounds(  # noqa: DOC503 (raises ApiClientError indirectly)
    self,
    component_id: int,
    lower: float,
    upper: float,
) -> None:
    """Send `SetBoundsParam`s received from a channel to the Microgrid service.

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

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

    target_metric = (
        microgrid_pb2.SetBoundsParam.TargetMetric.TARGET_METRIC_POWER_ACTIVE
    )
    await client.call_stub_method(
        self,
        lambda: self.stub.AddInclusionBounds(
            microgrid_pb2.SetBoundsParam(
                component_id=component_id,
                target_metric=target_metric,
                bounds=metrics_pb2.Bounds(lower=lower, upper=upper),
            ),
            timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
        ),
        method_name="AddInclusionBounds",
    )
set_power async ¤
set_power(component_id: int, power_w: float) -> None

Send request to the Microgrid to set power for component.

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

PARAMETER DESCRIPTION
component_id

id of the component to set power.

TYPE: int

power_w

power to set for the component.

TYPE: float

RAISES DESCRIPTION
ApiClientError

If the are any errors communicating with the Microgrid API, most likely a subclass of GrpcError.
Source code in frequenz/client/microgrid/_client.py 
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
async def set_power(  # noqa: DOC502 (raises ApiClientError indirectly)
    self, component_id: int, power_w: float
) -> None:
    """Send request to the Microgrid to set power for component.

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


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

    Raises:
        ApiClientError: If the are any errors communicating with the Microgrid API,
            most likely a subclass of
            [GrpcError][frequenz.client.microgrid.GrpcError].
    """
    await client.call_stub_method(
        self,
        lambda: self.stub.SetPowerActive(
            microgrid_pb2.SetPowerActiveParam(
                component_id=component_id, power=power_w
            ),
            timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
        ),
        method_name="SetPowerActive",
    )
set_reactive_power async ¤
set_reactive_power(
    component_id: int, reactive_power_var: float
) -> None

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

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

PARAMETER DESCRIPTION
component_id

id of the component to set power.

TYPE: int

reactive_power_var

reactive power to set for the component.

TYPE: float

RAISES DESCRIPTION
ApiClientError

If the are any errors communicating with the Microgrid API, most likely a subclass of GrpcError.
Source code in frequenz/client/microgrid/_client.py 
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
async def set_reactive_power(  # noqa: DOC502 (raises ApiClientError indirectly)
    self, component_id: int, reactive_power_var: float
) -> None:
    """Send request to the Microgrid to set reactive power for component.

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

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

    Raises:
        ApiClientError: If the are any errors communicating with the Microgrid API,
            most likely a subclass of
            [GrpcError][frequenz.client.microgrid.GrpcError].
    """
    await client.call_stub_method(
        self,
        lambda: self.stub.SetPowerReactive(
            microgrid_pb2.SetPowerReactiveParam(
                component_id=component_id, power=reactive_power_var
            ),
            timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
        ),
        method_name="SetPowerReactive",
    )

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