Index

frequenz.sdk.timeseries.battery_pool

Manage a pool of batteries.

Classes

frequenz.sdk.timeseries.battery_pool.BatteryPool

Bases: ComponentPool[BatteryPoolReferenceStore, BatteryPoolReport]

An interface for interaction with pools of batteries.

Provides

• properties for fetching reporting streams of instantaneous power, soc, capacity values and available power bounds and other status through power_status.
• control methods for proposing power values, namely: propose_power, propose_charge and propose_discharge.

Source code in src/frequenz/sdk/timeseries/battery_pool/_battery_pool.py

class BatteryPool(ComponentPool[BatteryPoolReferenceStore, BatteryPoolReport]):
    """An interface for interaction with pools of batteries.

    Provides:
      - properties for fetching reporting streams of instantaneous
        [power][frequenz.sdk.timeseries.battery_pool.BatteryPool.power],
        [soc][frequenz.sdk.timeseries.battery_pool.BatteryPool.soc],
        [capacity][frequenz.sdk.timeseries.battery_pool.BatteryPool.capacity] values and
        available power bounds and other status through
        [power_status][frequenz.sdk.timeseries.battery_pool.BatteryPool.power_status].
      - control methods for proposing power values, namely:
        [propose_power][frequenz.sdk.timeseries.battery_pool.BatteryPool.propose_power],
        [propose_charge][frequenz.sdk.timeseries.battery_pool.BatteryPool.propose_charge] and
        [propose_discharge][frequenz.sdk.timeseries.battery_pool.BatteryPool.propose_discharge].
    """

    async def propose_charge(self, power: Power | None) -> None:
        """Set the given charge power for the batteries in the pool.

        Power values need to be positive values, indicating charge power.

        When using the Passive Sign Convention, the
        [`propose_power`][frequenz.sdk.timeseries.battery_pool.BatteryPool.propose_power]
        method might be more convenient.

        If the same batteries are shared by multiple actors, the behaviour is the same
        as that of the `propose_power` method, when calling it with `None` bounds.  The
        bounds for lower priority actors can't be specified with this method.  If that's
        required, use the `propose_power` method instead.

        Args:
            power: The unsigned charge power to propose for the batteries in the pool.
                If None, the proposed power of higher priority actors will take
                precedence as the target power.

        Raises:
            ValueError: If the given power is negative.
        """
        if power and power < Power.zero():
            raise ValueError("Charge power must be positive.")
        await self._pool_ref_store.power_manager_requests_sender.send(
            _power_managing.Proposal(
                source_id=self._source_id,
                preferred_power=power,
                bounds=timeseries.Bounds(None, None),
                component_ids=self._pool_ref_store.component_ids,
                priority=self._priority,
                creation_time=asyncio.get_running_loop().time(),
            )
        )

    async def propose_discharge(self, power: Power | None) -> None:
        """Set the given discharge power for the batteries in the pool.

        Power values need to be positive values, indicating discharge power.

        When using the Passive Sign Convention, the
        [`propose_power`][frequenz.sdk.timeseries.battery_pool.BatteryPool.propose_power]
        method might be more convenient.

        If the same batteries are shared by multiple actors, the behaviour is the same
        as that of the `propose_power` method, when calling it with `None` bounds.  The
        bounds for lower priority actors can't be specified with this method.  If that's
        required, use the `propose_power` method instead.

        Args:
            power: The unsigned discharge power to propose for the batteries in the
                pool.  If None, the proposed power of higher priority actors will take
                precedence as the target power.

        Raises:
            ValueError: If the given power is negative.
        """
        if power:
            if power < Power.zero():
                raise ValueError("Discharge power must be positive.")
            power = -power
        await self._pool_ref_store.power_manager_requests_sender.send(
            _power_managing.Proposal(
                source_id=self._source_id,
                preferred_power=power,
                bounds=timeseries.Bounds(None, None),
                component_ids=self._pool_ref_store.component_ids,
                priority=self._priority,
                creation_time=asyncio.get_running_loop().time(),
            )
        )

    @property
    @override
    def power(self) -> Formula[Power]:
        """Fetch the total power of the batteries in the pool.

        This formula produces values that are in the Passive Sign Convention (PSC).

        If a formula to calculate this metric is not already running, it will be
        started.

        A receiver from the formula can be obtained by calling the `new_receiver`
        method.

        Returns:
            A Formula that will calculate and stream the total power of all
                batteries in the pool.
        """
        return self._pool_ref_store.formula_pool.from_power_formula(
            "battery_pool_power",
            connection_manager.get().component_graph.battery_formula(
                self._pool_ref_store.component_ids
            ),
        )

    @property
    def soc(self) -> ReceiverFetcher[Sample[Percentage]]:
        """Fetch the normalized average weighted-by-capacity SoC values for the pool.

        The SoC values are normalized to the 0-100% range and clamped if they are out
        of bounds. Only values from working batteries with operational inverters are
        considered in the calculation.

        Average SoC is calculated using the formula:
        ```
        working_batteries: Set[BatteryData] # working batteries from the battery pool

        soc_scaled = min(max(
            0,
            (soc - soc_lower_bound) / (soc_upper_bound - soc_lower_bound) * 100,
        ), 100)
        used_capacity = sum(
            battery.usable_capacity * battery.soc_scaled
            for battery in working_batteries
        )
        total_capacity = sum(battery.usable_capacity for battery in working_batteries)
        average_soc = used_capacity/total_capacity
        ```

        `None` values will be sent if there are no working batteries with operational
        inverters to calculate the metric with.

        A receiver from the MetricAggregator can be obtained by calling the
        `new_receiver` method.

        Returns:
            A MetricAggregator that will calculate and stream the aggregate SoC of all
                batteries in the pool, considering only working batteries with
                operational inverters.
        """
        assert isinstance(self._pool_ref_store, BatteryPoolReferenceStore)

        method_name = SendOnUpdate.name() + "_" + SoCCalculator.name()

        if method_name not in self._pool_ref_store._active_methods:
            calculator = SoCCalculator(self._pool_ref_store.component_ids)
            self._pool_ref_store._active_methods[method_name] = SendOnUpdate(
                metric_calculator=calculator,
                working_batteries=self._pool_ref_store._working_batteries,
                min_update_interval=self._pool_ref_store._min_update_interval,
            )

        return self._pool_ref_store._active_methods[method_name]

    @property
    def temperature(self) -> ReceiverFetcher[Sample[Temperature]]:
        """Fetch the average temperature of the batteries in the pool.

        Returns:
            A MetricAggregator that will calculate and stream the average temperature
                of all batteries in the pool.
        """
        assert isinstance(self._pool_ref_store, BatteryPoolReferenceStore)

        method_name = SendOnUpdate.name() + "_" + TemperatureCalculator.name()

        if method_name not in self._pool_ref_store._active_methods:
            calculator = TemperatureCalculator(self._pool_ref_store.component_ids)
            self._pool_ref_store._active_methods[method_name] = SendOnUpdate(
                metric_calculator=calculator,
                working_batteries=self._pool_ref_store._working_batteries,
                min_update_interval=self._pool_ref_store._min_update_interval,
            )
        return self._pool_ref_store._active_methods[method_name]

    @property
    def capacity(self) -> ReceiverFetcher[Sample[Energy]]:
        """Get a receiver to receive new capacity metrics when they change.

        The reported capacity values consider only working batteries with operational
        inverters.

        Calculated with the formula:
        ```
        working_batteries: Set[BatteryData] # working batteries from the battery pool
        total_capacity = sum(
            battery.capacity * (soc_upper_bound - soc_lower_bound) / 100
            for battery in working_batteries
        )
        ```

        `None` will be sent if there are no working batteries with operational
        inverters to calculate metrics.

        A receiver from the MetricAggregator can be obtained by calling the
        `new_receiver` method.

        Returns:
            A MetricAggregator that will calculate and stream the capacity of all
                batteries in the pool, considering only working batteries with
                operational inverters.
        """
        assert isinstance(self._pool_ref_store, BatteryPoolReferenceStore)

        method_name = SendOnUpdate.name() + "_" + CapacityCalculator.name()

        if method_name not in self._pool_ref_store._active_methods:
            calculator = CapacityCalculator(self._pool_ref_store.component_ids)
            self._pool_ref_store._active_methods[method_name] = SendOnUpdate(
                metric_calculator=calculator,
                working_batteries=self._pool_ref_store._working_batteries,
                min_update_interval=self._pool_ref_store._min_update_interval,
            )

        return self._pool_ref_store._active_methods[method_name]

    @override
    @property
    def system_power_bounds(self) -> ReceiverFetcher[SystemBounds]:
        """Get receiver to receive new power bounds when they change.

        Power bounds refer to the min and max power that a battery can
        discharge or charge at and is also denoted as SoP.

        Power bounds formulas are described in the receiver return type.
        None will be sent if there is no component to calculate metrics.

        A receiver from the MetricAggregator can be obtained by calling the
        `new_receiver` method.

        Returns:
            A MetricAggregator that will calculate and stream the power bounds
                of all batteries in the pool.
        """
        assert isinstance(self._pool_ref_store, BatteryPoolReferenceStore)

        method_name = SendOnUpdate.name() + "_" + PowerBoundsCalculator.name()

        if method_name not in self._pool_ref_store._active_methods:
            calculator = PowerBoundsCalculator(self._pool_ref_store.component_ids)
            self._pool_ref_store._active_methods[method_name] = SendOnUpdate(
                metric_calculator=calculator,
                working_batteries=self._pool_ref_store._working_batteries,
                min_update_interval=self._pool_ref_store._min_update_interval,
            )

        return self._pool_ref_store._active_methods[method_name]

Attributes

capacity property

capacity: ReceiverFetcher[Sample[Energy]]

Get a receiver to receive new capacity metrics when they change.

The reported capacity values consider only working batteries with operational inverters.

Calculated with the formula:

working_batteries: Set[BatteryData] # working batteries from the battery pool
total_capacity = sum(
    battery.capacity * (soc_upper_bound - soc_lower_bound) / 100
    for battery in working_batteries
)

None will be sent if there are no working batteries with operational inverters to calculate metrics.

A receiver from the MetricAggregator can be obtained by calling the new_receiver method.

RETURNS	DESCRIPTION
ReceiverFetcher[Sample[Energy]]	A MetricAggregator that will calculate and stream the capacity of all batteries in the pool, considering only working batteries with operational inverters.

component_ids property

component_ids: Set[ComponentId]

Return component IDs of all component IDs managed by this pool.

RETURNS	DESCRIPTION
Set[ComponentId]	Set of managed component IDs.

power property

power: Formula[Power]

Fetch the total power of the batteries in the pool.

This formula produces values that are in the Passive Sign Convention (PSC).

If a formula to calculate this metric is not already running, it will be started.

A receiver from the formula can be obtained by calling the new_receiver method.

RETURNS	DESCRIPTION
Formula[Power]	A Formula that will calculate and stream the total power of all batteries in the pool.

power_distribution_results property

power_distribution_results: ReceiverFetcher[Result]

Get a receiver to receive power distribution results.

RETURNS	DESCRIPTION
ReceiverFetcher[Result]	A receiver that will stream power distribution results for the pool's set of
ReceiverFetcher[Result]	components.

power_status property

power_status: ReceiverFetcher[ReportT]

Get a receiver to receive new power status reports when they change.

These include - the current inclusion/exclusion bounds available for the pool's priority, - the current target power for the pool's set of components, - the result of the last distribution request for the pool's set of components,.

RETURNS	DESCRIPTION
ReceiverFetcher[ReportT]	A receiver that will stream power status reports for the pool's priority.

soc property

soc: ReceiverFetcher[Sample[Percentage]]

Fetch the normalized average weighted-by-capacity SoC values for the pool.

The SoC values are normalized to the 0-100% range and clamped if they are out of bounds. Only values from working batteries with operational inverters are considered in the calculation.

Average SoC is calculated using the formula:

working_batteries: Set[BatteryData] # working batteries from the battery pool

soc_scaled = min(max(
    0,
    (soc - soc_lower_bound) / (soc_upper_bound - soc_lower_bound) * 100,
), 100)
used_capacity = sum(
    battery.usable_capacity * battery.soc_scaled
    for battery in working_batteries
)
total_capacity = sum(battery.usable_capacity for battery in working_batteries)
average_soc = used_capacity/total_capacity

None values will be sent if there are no working batteries with operational inverters to calculate the metric with.

A receiver from the MetricAggregator can be obtained by calling the new_receiver method.

RETURNS	DESCRIPTION
ReceiverFetcher[Sample[Percentage]]	A MetricAggregator that will calculate and stream the aggregate SoC of all batteries in the pool, considering only working batteries with operational inverters.

system_power_bounds property

system_power_bounds: ReceiverFetcher[SystemBounds]

Get receiver to receive new power bounds when they change.

Power bounds refer to the min and max power that a battery can discharge or charge at and is also denoted as SoP.

Power bounds formulas are described in the receiver return type. None will be sent if there is no component to calculate metrics.

A receiver from the MetricAggregator can be obtained by calling the new_receiver method.

RETURNS	DESCRIPTION
ReceiverFetcher[SystemBounds]	A MetricAggregator that will calculate and stream the power bounds of all batteries in the pool.

temperature property

temperature: ReceiverFetcher[Sample[Temperature]]

Fetch the average temperature of the batteries in the pool.

RETURNS	DESCRIPTION
ReceiverFetcher[Sample[Temperature]]	A MetricAggregator that will calculate and stream the average temperature of all batteries in the pool.

Functions

__init__

__init__(
    *,
    pool_ref_store: RefStoreT,
    name: str | None,
    priority: int
) -> None

Create an AbstractPool instance.

PARAMETER	DESCRIPTION
pool_ref_store	The pool reference store instance. TYPE: RefStoreT
name	An optional name used to identify this instance of the pool or a corresponding actor in the logs. TYPE: str | None
priority	The priority of the actor using this wrapper. TYPE: int

Source code in src/frequenz/sdk/timeseries/component_pool/_component_pool.py

def __init__(  # pylint: disable=too-many-arguments
    self,
    *,
    pool_ref_store: RefStoreT,
    name: str | None,
    priority: int,
) -> None:
    """Create an `AbstractPool` instance.

    Args:
        pool_ref_store: The pool reference store instance.
        name: An optional name used to identify this instance of the pool or a
            corresponding actor in the logs.
        priority: The priority of the actor using this wrapper.
    """
    self._pool_ref_store = pool_ref_store
    unique_id = str(uuid.uuid4())
    self._source_id = unique_id if name is None else f"{name}-{unique_id}"
    self._priority = priority

propose_charge async

propose_charge(power: Power | None) -> None

Set the given charge power for the batteries in the pool.

Power values need to be positive values, indicating charge power.

When using the Passive Sign Convention, the propose_power method might be more convenient.

If the same batteries are shared by multiple actors, the behaviour is the same as that of the propose_power method, when calling it with None bounds. The bounds for lower priority actors can't be specified with this method. If that's required, use the propose_power method instead.

PARAMETER	DESCRIPTION
power	The unsigned charge power to propose for the batteries in the pool. If None, the proposed power of higher priority actors will take precedence as the target power. TYPE: Power | None

RAISES	DESCRIPTION
ValueError	If the given power is negative.

Source code in src/frequenz/sdk/timeseries/battery_pool/_battery_pool.py

async def propose_charge(self, power: Power | None) -> None:
    """Set the given charge power for the batteries in the pool.

    Power values need to be positive values, indicating charge power.

    When using the Passive Sign Convention, the
    [`propose_power`][frequenz.sdk.timeseries.battery_pool.BatteryPool.propose_power]
    method might be more convenient.

    If the same batteries are shared by multiple actors, the behaviour is the same
    as that of the `propose_power` method, when calling it with `None` bounds.  The
    bounds for lower priority actors can't be specified with this method.  If that's
    required, use the `propose_power` method instead.

    Args:
        power: The unsigned charge power to propose for the batteries in the pool.
            If None, the proposed power of higher priority actors will take
            precedence as the target power.

    Raises:
        ValueError: If the given power is negative.
    """
    if power and power < Power.zero():
        raise ValueError("Charge power must be positive.")
    await self._pool_ref_store.power_manager_requests_sender.send(
        _power_managing.Proposal(
            source_id=self._source_id,
            preferred_power=power,
            bounds=timeseries.Bounds(None, None),
            component_ids=self._pool_ref_store.component_ids,
            priority=self._priority,
            creation_time=asyncio.get_running_loop().time(),
        )
    )

propose_discharge async

propose_discharge(power: Power | None) -> None

Set the given discharge power for the batteries in the pool.

Power values need to be positive values, indicating discharge power.

When using the Passive Sign Convention, the propose_power method might be more convenient.

If the same batteries are shared by multiple actors, the behaviour is the same as that of the propose_power method, when calling it with None bounds. The bounds for lower priority actors can't be specified with this method. If that's required, use the propose_power method instead.

PARAMETER	DESCRIPTION
power	The unsigned discharge power to propose for the batteries in the pool. If None, the proposed power of higher priority actors will take precedence as the target power. TYPE: Power | None

RAISES	DESCRIPTION
ValueError	If the given power is negative.

Source code in src/frequenz/sdk/timeseries/battery_pool/_battery_pool.py

async def propose_discharge(self, power: Power | None) -> None:
    """Set the given discharge power for the batteries in the pool.

    Power values need to be positive values, indicating discharge power.

    When using the Passive Sign Convention, the
    [`propose_power`][frequenz.sdk.timeseries.battery_pool.BatteryPool.propose_power]
    method might be more convenient.

    If the same batteries are shared by multiple actors, the behaviour is the same
    as that of the `propose_power` method, when calling it with `None` bounds.  The
    bounds for lower priority actors can't be specified with this method.  If that's
    required, use the `propose_power` method instead.

    Args:
        power: The unsigned discharge power to propose for the batteries in the
            pool.  If None, the proposed power of higher priority actors will take
            precedence as the target power.

    Raises:
        ValueError: If the given power is negative.
    """
    if power:
        if power < Power.zero():
            raise ValueError("Discharge power must be positive.")
        power = -power
    await self._pool_ref_store.power_manager_requests_sender.send(
        _power_managing.Proposal(
            source_id=self._source_id,
            preferred_power=power,
            bounds=timeseries.Bounds(None, None),
            component_ids=self._pool_ref_store.component_ids,
            priority=self._priority,
            creation_time=asyncio.get_running_loop().time(),
        )
    )

propose_power async

propose_power(
    power: Power | None,
    bounds: Bounds[Power | None] = Bounds(None, None),
) -> None

Send a proposal to the power manager for the pool's underlying components.

This proposal is for the maximum power that can be set for the components in the pool. The actual production or consumption might be lower.

Details on how the power manager handles proposals can be found in the Microgrid documentation.

PARAMETER	DESCRIPTION
power	The power to propose. If None, this proposal will not have any effect on the target power, unless bounds are specified. When specified without bounds, bounds for lower priority actors will be shifted by this power. If both are None, it is equivalent to not having a proposal or withdrawing a previous one. TYPE: Power | None
bounds	The power bounds for the proposal. When specified, these bounds will limit the bounds for lower priority actors. TYPE: Bounds[Power | None] DEFAULT: Bounds(None, None)

Source code in src/frequenz/sdk/timeseries/component_pool/_component_pool.py

async def propose_power(
    self,
    power: Power | None,
    bounds: Bounds[Power | None] = Bounds(None, None),
) -> None:
    """Send a proposal to the power manager for the pool's underlying components.

    This proposal is for the maximum power that can be set for the components in
    the pool. The actual production or consumption might be lower.

    Details on how the power manager handles proposals can be found in the
    [Microgrid][frequenz.sdk.microgrid--setting-power] documentation.

    Args:
        power: The power to propose.  If `None`,
            this proposal will not have any effect on the target power, unless
            bounds are specified.  When specified without bounds, bounds for lower
            priority actors will be shifted by this power.  If both are `None`, it
            is equivalent to not having a proposal or withdrawing a previous one.
        bounds: The power bounds for the proposal. When specified, these bounds will
            limit the bounds for lower priority actors.
    """
    await self._pool_ref_store.power_manager_requests_sender.send(
        _power_managing.Proposal(
            source_id=self._source_id,
            preferred_power=power,
            bounds=bounds,
            component_ids=self._pool_ref_store.component_ids,
            priority=self._priority,
            creation_time=asyncio.get_running_loop().time(),
        )
    )

stop async

stop() -> None

Stop all tasks and channels owned by the pool.

Source code in src/frequenz/sdk/timeseries/component_pool/_component_pool.py

async def stop(self) -> None:
    """Stop all tasks and channels owned by the pool."""

frequenz.sdk.timeseries.battery_pool.BatteryPoolReport

Bases: ComponentPoolReport, Protocol

A status report for a battery pool.

Source code in src/frequenz/sdk/timeseries/battery_pool/messages.py

class BatteryPoolReport(ComponentPoolReport, 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 src/frequenz/sdk/timeseries/battery_pool/messages.py

@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.
    """