Skip to content

messages

frequenz.sdk.timeseries.battery_pool.messages ¤

Types for exposing battery pool reports.

Attributes¤

frequenz.sdk.timeseries.battery_pool.messages.Result module-attribute ¤

Result = Success | PartialFailure | Error | OutOfBounds

Power distribution result.

Example: Handling power distribution results

```python
from typing import assert_never

from frequenz.sdk.actor.power_distributing import (
    Error,
    OutOfBounds,
    PartialFailure,
    Result,
    Success,
)
from frequenz.sdk.actor.power_distributing.request import Request
from frequenz.sdk.actor.power_distributing.result import PowerBounds
from frequenz.sdk.timeseries._quantities import Power

def handle_power_request_result(result: Result) -> None:
    match result:
        case Success() as success:
            print(f"Power request was successful: {success}")
        case PartialFailure() as partial_failure:
            print(f"Power request was partially successful: {partial_failure}")
        case OutOfBounds() as out_of_bounds:
            print(f"Power request was out of bounds: {out_of_bounds}")
        case Error() as error:
            print(f"Power request failed: {error}")
        case _ as unreachable:
            assert_never(unreachable)

request = Request(
    namespace="TestChannel",
    power=Power.from_watts(123.4),
    component_ids={8, 18},
)

results: list[Result] = [
    Success(
        request,
        succeeded_power=Power.from_watts(123.4),
        succeeded_components={8, 18},
        excess_power=Power.zero(),
    ),
    PartialFailure(
        request,
        succeeded_power=Power.from_watts(103.4),
        succeeded_components={8},
        excess_power=Power.zero(),
        failed_components={18},
        failed_power=Power.from_watts(20.0),
    ),
    OutOfBounds(request, bounds=PowerBounds(0, 0, 0, 800)),
    Error(request, msg="The components are not available"),
]

for r in results:
    handle_power_request_result(r)
```

Classes¤

frequenz.sdk.timeseries.battery_pool.messages.BatteryPoolReport ¤

Bases: Protocol

A status report for a battery pool.

Source code in frequenz/sdk/timeseries/battery_pool/messages.py 
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
class BatteryPoolReport(typing.Protocol):
    """A status report for a battery pool."""

    @property
    def target_power(self) -> Power | None:
        """The currently set power for the batteries."""

    @property
    def bounds(self) -> Bounds[Power] | None:
        """The usable bounds for the batteries.

        These bounds are adjusted to any restrictions placed by actors with higher
        priorities.

        There might be exclusion zones within these bounds. If necessary, the
        [`adjust_to_bounds`][frequenz.sdk.timeseries.battery_pool.messages.BatteryPoolReport.adjust_to_bounds]
        method may be used to check if a desired power value fits the bounds, or to get
        the closest possible power values that do fit the bounds.
        """

    @abc.abstractmethod
    def adjust_to_bounds(self, power: Power) -> tuple[Power | None, Power | None]:
        """Adjust a power value to the bounds.

        This method can be used to adjust a desired power value to the power bounds
        available to the actor.

        If the given power value falls within the usable bounds, it will be returned
        unchanged.

        If it falls outside the usable bounds, the closest possible value on the
        corresponding side will be returned.  For example, if the given power is lower
        than the lowest usable power, only the lowest usable power will be returned, and
        similarly for the highest usable power.

        If the given power falls within an exclusion zone that's contained within the
        usable bounds, the closest possible power values on both sides will be returned.

        !!! note
            It is completely optional to use this method to adjust power values before
            proposing them, because the battery pool will do this automatically.  This
            method is provided for convenience, and for granular control when there are
            two possible power values, both of which fall within the available bounds.

        Example:
            ```python
            from frequenz.sdk import microgrid

            power_status_rx = microgrid.new_battery_pool(
                priority=5,
            ).power_status.new_receiver()
            power_status = await power_status_rx.receive()
            desired_power = Power.from_watts(1000.0)

            match power_status.adjust_to_bounds(desired_power):
                case (power, _) if power == desired_power:
                    print("Desired power is available.")
                case (None, power) | (power, None) if power:
                    print(f"Closest available power is {power}.")
                case (lower, upper) if lower and upper:
                    print(f"Two options {lower}, {upper} to propose to battery pool.")
                case (None, None):
                    print("No available power")
            ```

        Args:
            power: The power value to adjust.

        Returns:
            A tuple of the closest power values to the desired power that fall within
                the available bounds for the actor.
        """
Attributes¤
bounds property ¤
bounds: Bounds[Power] | None

The usable bounds for the batteries.

These bounds are adjusted to any restrictions placed by actors with higher priorities.

There might be exclusion zones within these bounds. If necessary, the adjust_to_bounds method may be used to check if a desired power value fits the bounds, or to get the closest possible power values that do fit the bounds.

target_power property ¤
target_power: Power | None

The currently set power for the batteries.

Functions¤
adjust_to_bounds abstractmethod ¤
adjust_to_bounds(
    power: Power,
) -> tuple[Power | None, Power | None]

Adjust a power value to the bounds.

This method can be used to adjust a desired power value to the power bounds available to the actor.

If the given power value falls within the usable bounds, it will be returned unchanged.

If it falls outside the usable bounds, the closest possible value on the corresponding side will be returned. For example, if the given power is lower than the lowest usable power, only the lowest usable power will be returned, and similarly for the highest usable power.

If the given power falls within an exclusion zone that's contained within the usable bounds, the closest possible power values on both sides will be returned.

Note

It is completely optional to use this method to adjust power values before proposing them, because the battery pool will do this automatically. This method is provided for convenience, and for granular control when there are two possible power values, both of which fall within the available bounds.

Example 
from frequenz.sdk import microgrid

power_status_rx = microgrid.new_battery_pool(
    priority=5,
).power_status.new_receiver()
power_status = await power_status_rx.receive()
desired_power = Power.from_watts(1000.0)

match power_status.adjust_to_bounds(desired_power):
    case (power, _) if power == desired_power:
        print("Desired power is available.")
    case (None, power) | (power, None) if power:
        print(f"Closest available power is {power}.")
    case (lower, upper) if lower and upper:
        print(f"Two options {lower}, {upper} to propose to battery pool.")
    case (None, None):
        print("No available power")
PARAMETER DESCRIPTION
power

The power value to adjust.

TYPE: Power

RETURNS DESCRIPTION
tuple[Power | None, Power | None]

A tuple of the closest power values to the desired power that fall within the available bounds for the actor.
Source code in frequenz/sdk/timeseries/battery_pool/messages.py 
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
@abc.abstractmethod
def adjust_to_bounds(self, power: Power) -> tuple[Power | None, Power | None]:
    """Adjust a power value to the bounds.

    This method can be used to adjust a desired power value to the power bounds
    available to the actor.

    If the given power value falls within the usable bounds, it will be returned
    unchanged.

    If it falls outside the usable bounds, the closest possible value on the
    corresponding side will be returned.  For example, if the given power is lower
    than the lowest usable power, only the lowest usable power will be returned, and
    similarly for the highest usable power.

    If the given power falls within an exclusion zone that's contained within the
    usable bounds, the closest possible power values on both sides will be returned.

    !!! note
        It is completely optional to use this method to adjust power values before
        proposing them, because the battery pool will do this automatically.  This
        method is provided for convenience, and for granular control when there are
        two possible power values, both of which fall within the available bounds.

    Example:
        ```python
        from frequenz.sdk import microgrid

        power_status_rx = microgrid.new_battery_pool(
            priority=5,
        ).power_status.new_receiver()
        power_status = await power_status_rx.receive()
        desired_power = Power.from_watts(1000.0)

        match power_status.adjust_to_bounds(desired_power):
            case (power, _) if power == desired_power:
                print("Desired power is available.")
            case (None, power) | (power, None) if power:
                print(f"Closest available power is {power}.")
            case (lower, upper) if lower and upper:
                print(f"Two options {lower}, {upper} to propose to battery pool.")
            case (None, None):
                print("No available power")
        ```

    Args:
        power: The power value to adjust.

    Returns:
        A tuple of the closest power values to the desired power that fall within
            the available bounds for the actor.
    """

frequenz.sdk.timeseries.battery_pool.messages.Error dataclass ¤

Bases: _BaseResultMixin

Result returned when an error occurred and power was not set at all.

Source code in frequenz/sdk/microgrid/_power_distributing/result.py 
62
63
64
65
66
67
@dataclasses.dataclass
class Error(_BaseResultMixin):
    """Result returned when an error occurred and power was not set at all."""

    msg: str
    """The error message explaining why error happened."""
Attributes¤
msg instance-attribute ¤
msg: str

The error message explaining why error happened.

request instance-attribute ¤
request: Request

The user's request to which this message responds.

frequenz.sdk.timeseries.battery_pool.messages.OutOfBounds dataclass ¤

Bases: _BaseResultMixin

Result returned when the power was not set because it was out of bounds.

This result happens when the originating request was done with adjust_power = False and the requested power is not within the available bounds.

Source code in frequenz/sdk/microgrid/_power_distributing/result.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@dataclasses.dataclass
class OutOfBounds(_BaseResultMixin):
    """Result returned when the power was not set because it was out of bounds.

    This result happens when the originating request was done with
    `adjust_power = False` and the requested power is not within the available bounds.
    """

    bounds: PowerBounds
    """The power bounds for the requested components.

    If the requested power negative, then this value is the lower bound.
    Otherwise it is upper bound.
    """
Attributes¤
bounds instance-attribute ¤
bounds: PowerBounds

The power bounds for the requested components.

If the requested power negative, then this value is the lower bound. Otherwise it is upper bound.

request instance-attribute ¤
request: Request

The user's request to which this message responds.

frequenz.sdk.timeseries.battery_pool.messages.PartialFailure dataclass ¤

Bases: _BaseSuccessMixin, _BaseResultMixin

Result returned when some of the components had an error setting the power.

Source code in frequenz/sdk/microgrid/_power_distributing/result.py 
51
52
53
54
55
56
57
58
59
@dataclasses.dataclass
class PartialFailure(_BaseSuccessMixin, _BaseResultMixin):
    """Result returned when some of the components had an error setting the power."""

    failed_power: Power
    """The part of the requested power that failed to be set."""

    failed_components: abc.Set[int]
    """The subset of batteries for which the request failed."""
Attributes¤
excess_power instance-attribute ¤
excess_power: Power

The part of the requested power that could not be fulfilled.

This happens when the requested power is outside the available power bounds.

failed_components instance-attribute ¤
failed_components: Set[int]

The subset of batteries for which the request failed.

failed_power instance-attribute ¤
failed_power: Power

The part of the requested power that failed to be set.

request instance-attribute ¤
request: Request

The user's request to which this message responds.

succeeded_components instance-attribute ¤
succeeded_components: Set[int]

The subset of components for which power was set successfully.

succeeded_power instance-attribute ¤
succeeded_power: Power

The part of the requested power that was successfully set.

frequenz.sdk.timeseries.battery_pool.messages.Success dataclass ¤

Bases: _BaseSuccessMixin, _BaseResultMixin

Result returned when setting the power was successful for all components.

Source code in frequenz/sdk/microgrid/_power_distributing/result.py 
46
47
48
@dataclasses.dataclass
class Success(_BaseSuccessMixin, _BaseResultMixin):  # Order matters here. See above.
    """Result returned when setting the power was successful for all components."""
Attributes¤
excess_power instance-attribute ¤
excess_power: Power

The part of the requested power that could not be fulfilled.

This happens when the requested power is outside the available power bounds.

request instance-attribute ¤
request: Request

The user's request to which this message responds.

succeeded_components instance-attribute ¤
succeeded_components: Set[int]

The subset of components for which power was set successfully.

succeeded_power instance-attribute ¤
succeeded_power: Power

The part of the requested power that was successfully set.