Skip to content

Make PowerManagingActor more modular and generic #773

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 11 commits into from
Nov 9, 2023
Merged
Show file tree
Hide file tree
Changes from 6 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 31 additions & 52 deletions src/frequenz/sdk/actor/_power_managing/_base_classes.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
from . import _bounds

if typing.TYPE_CHECKING:
from ...timeseries.battery_pool import PowerMetrics
from ...timeseries._base_types import SystemBounds
from .. import power_distributing


Expand All @@ -27,8 +27,8 @@ class ReportRequest:
source_id: str
"""The source ID of the actor sending the request."""

battery_ids: frozenset[int]
"""The battery IDs to report on."""
component_ids: frozenset[int]
"""The component IDs to report on."""

priority: int
"""The priority of the actor ."""
Expand All @@ -40,15 +40,15 @@ def get_channel_name(self) -> str:
The channel name to use to identify the corresponding report channel
from the channel registry.
"""
return f"power_manager.report.{self.battery_ids=}.{self.priority=}"
return f"power_manager.report.{self.component_ids=}.{self.priority=}"


@dataclasses.dataclass(frozen=True, kw_only=True)
class Report:
"""Current PowerManager report for a set of batteries."""
"""Current PowerManager report for a set of components."""

target_power: Power | None
"""The currently set power for the batteries."""
"""The currently set power for the components."""

distribution_result: power_distributing.Result | None
"""The result of the last power distribution.
Expand All @@ -57,14 +57,14 @@ class Report:
"""

_inclusion_bounds: timeseries.Bounds[Power] | None
"""The available inclusion bounds for the batteries, for the actor's priority.
"""The available inclusion bounds for the components, for the actor's priority.

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

_exclusion_bounds: timeseries.Bounds[Power] | None
"""The exclusion bounds for the batteries.
"""The exclusion bounds for the components.

The power manager doesn't manage exclusion bounds, so these are aggregations of
values reported by the microgrid API.
Expand All @@ -75,15 +75,14 @@ class Report:

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

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.Report.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.
`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.
"""
return self._inclusion_bounds

Expand All @@ -106,29 +105,9 @@ def adjust_to_bounds(self, power: Power) -> tuple[Power | None, Power | None]:

!!! note
It is completely optional to use this method to adjust power values before
proposing them through the battery pool, 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.battery_pool().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")
```
proposing them, because the PowerManager 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.

Args:
power: The power value to adjust.
Expand All @@ -150,13 +129,13 @@ def adjust_to_bounds(self, power: Power) -> tuple[Power | None, Power | None]:

@dataclasses.dataclass(frozen=True, kw_only=True)
class Proposal:
"""A proposal for a battery to be charged or discharged."""
"""A proposal for a set of components to be charged or discharged."""

source_id: str
"""The source ID of the actor sending the request."""

preferred_power: Power | None
"""The preferred power to be distributed to the batteries.
"""The preferred power to be distributed to the components.

If `None`, the preferred power of higher priority actors will get precedence.
"""
Expand All @@ -166,12 +145,12 @@ class Proposal:

These bounds will apply to actors with a lower priority, and can be overridden by
bounds from actors with a higher priority. If None, the power bounds will be set to
the maximum power of the batteries in the pool. This is currently an experimental
the maximum power of the components in the pool. This is currently an experimental
feature.
"""

battery_ids: frozenset[int]
"""The battery IDs to distribute the power to."""
component_ids: frozenset[int]
"""The component IDs to distribute the power to."""

priority: int
"""The priority of the actor sending the proposal."""
Expand Down Expand Up @@ -207,23 +186,23 @@ class BaseAlgorithm(abc.ABC):
@abc.abstractmethod
def calculate_target_power(
self,
battery_ids: frozenset[int],
component_ids: frozenset[int],
proposal: Proposal | None,
system_bounds: PowerMetrics,
system_bounds: SystemBounds,
must_return_power: bool = False,
) -> Power | None:
"""Calculate and return the target power for the given batteries.
"""Calculate and return the target power for the given components.

Args:
battery_ids: The battery IDs to calculate the target power for.
component_ids: The component IDs to calculate the target power for.
proposal: If given, the proposal to added to the bucket, before the target
power is calculated.
system_bounds: The system bounds for the batteries in the proposal.
system_bounds: The system bounds for the components in the proposal.
must_return_power: If `True`, the algorithm must return a target power,
even if it hasn't changed since the last call.

Returns:
The new target power for the batteries, or `None` if the target power
The new target power for the components, or `None` if the target power
didn't change.
"""

Expand All @@ -232,19 +211,19 @@ def calculate_target_power(
@abc.abstractmethod
def get_status(
self,
battery_ids: frozenset[int],
component_ids: frozenset[int],
priority: int,
system_bounds: PowerMetrics,
system_bounds: SystemBounds,
distribution_result: power_distributing.Result | None,
) -> Report:
"""Get the bounds for a set of batteries, for the given priority.
"""Get the bounds for a set of components, for the given priority.

Args:
battery_ids: The IDs of the batteries to get the bounds for.
component_ids: The IDs of the components to get the bounds for.
priority: The priority of the actor for which the bounds are requested.
system_bounds: The system bounds for the batteries.
system_bounds: The system bounds for the components.
distribution_result: The result of the last power distribution.

Returns:
The bounds for the batteries.
The bounds for the components.
"""
Loading