Index

frequenz.sdk.timeseries.battery_pool

Manage a pool of batteries.

Classes

frequenz.sdk.timeseries.battery_pool.BatteryPool

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 frequenz/sdk/timeseries/battery_pool/_battery_pool.py

class BatteryPool:
    """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].
    """

    def __init__(
        self,
        *,
        pool_ref_store: BatteryPoolReferenceStore,
        name: str | None,
        priority: int,
    ):
        """Create a BatteryPool instance.

        !!! note
            `BatteryPool` instances are not meant to be created directly by users.  Use
            the [`microgrid.new_battery_pool`][frequenz.sdk.microgrid.new_battery_pool]
            method for creating `BatteryPool` instances.

        Args:
            pool_ref_store: The battery 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

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

        Power values need to follow the Passive Sign Convention (PSC). That is, positive
        values indicate charge power and negative values indicate discharge power.

        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 for the batteries in the pool.  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, this 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._batteries,
                priority=self._priority,
                creation_time=asyncio.get_running_loop().time(),
            )
        )

    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._batteries,
                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._batteries,
                priority=self._priority,
                creation_time=asyncio.get_running_loop().time(),
            )
        )

    @property
    def component_ids(self) -> abc.Set[int]:
        """Return ids of the batteries in the pool.

        Returns:
            Ids of the batteries in the pool
        """
        return self._pool_ref_store._batteries

    @property
    def power(self) -> FormulaEngine[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 engine to calculate this metric is not already running, it will be
        started.

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

        Returns:
            A FormulaEngine that will calculate and stream the total power of all
                batteries in the pool.
        """
        engine = self._pool_ref_store._formula_pool.from_power_formula_generator(
            "battery_pool_power",
            BatteryPowerFormula,
            FormulaGeneratorConfig(
                component_ids=self._pool_ref_store._batteries,
                allow_fallback=True,
            ),
        )
        assert isinstance(engine, FormulaEngine)
        return engine

    @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.
        """
        method_name = SendOnUpdate.name() + "_" + SoCCalculator.name()

        if method_name not in self._pool_ref_store._active_methods:
            calculator = SoCCalculator(self._pool_ref_store._batteries)
            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.
        """
        method_name = SendOnUpdate.name() + "_" + TemperatureCalculator.name()
        if method_name not in self._pool_ref_store._active_methods:
            calculator = TemperatureCalculator(self._pool_ref_store._batteries)
            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.
        """
        method_name = SendOnUpdate.name() + "_" + CapacityCalculator.name()

        if method_name not in self._pool_ref_store._active_methods:
            calculator = CapacityCalculator(self._pool_ref_store._batteries)
            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 power_status(self) -> ReceiverFetcher[BatteryPoolReport]:
        """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 batteries,
          - the result of the last distribution request for the pool's set of batteries.

        Returns:
            A receiver that will stream power status reports for the pool's priority.
        """
        sub = _power_managing.ReportRequest(
            source_id=self._source_id,
            priority=self._priority,
            component_ids=self._pool_ref_store._batteries,
        )
        self._pool_ref_store._power_bounds_subs[sub.get_channel_name()] = (
            asyncio.create_task(
                self._pool_ref_store._power_manager_bounds_subscription_sender.send(sub)
            )
        )
        channel = self._pool_ref_store._channel_registry.get_or_create(
            _power_managing._Report, sub.get_channel_name()
        )
        channel.resend_latest = True

        return channel

    @property
    def power_distribution_results(self) -> ReceiverFetcher[_power_distributing.Result]:
        """Get a receiver to receive power distribution results.

        Returns:
            A receiver that will stream power distribution results for the pool's set of
            batteries.
        """
        return MappingReceiverFetcher(
            self._pool_ref_store._power_dist_results_fetcher,
            lambda recv: recv.filter(
                lambda x: x.request.component_ids == self._pool_ref_store._batteries
            ),
        )

    @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 send 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.
        """
        method_name = SendOnUpdate.name() + "_" + PowerBoundsCalculator.name()

        if method_name not in self._pool_ref_store._active_methods:
            calculator = PowerBoundsCalculator(self._pool_ref_store._batteries)
            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]

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

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[int]

Return ids of the batteries in the pool.

RETURNS	DESCRIPTION
Set[int]	Ids of the batteries in the pool

power property

power: FormulaEngine[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 engine to calculate this metric is not already running, it will be started.

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

RETURNS	DESCRIPTION
FormulaEngine[Power]	A FormulaEngine 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]	batteries.

power_status property

power_status: ReceiverFetcher[BatteryPoolReport]

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 batteries, - the result of the last distribution request for the pool's set of batteries.

RETURNS	DESCRIPTION
ReceiverFetcher[BatteryPoolReport]	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.

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: BatteryPoolReferenceStore,
    name: str | None,
    priority: int
)

Create a BatteryPool instance.

Note

BatteryPool instances are not meant to be created directly by users. Use the microgrid.new_battery_pool method for creating BatteryPool instances.

PARAMETER	DESCRIPTION
pool_ref_store	The battery pool reference store instance. TYPE: BatteryPoolReferenceStore
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 frequenz/sdk/timeseries/battery_pool/_battery_pool.py

def __init__(
    self,
    *,
    pool_ref_store: BatteryPoolReferenceStore,
    name: str | None,
    priority: int,
):
    """Create a BatteryPool instance.

    !!! note
        `BatteryPool` instances are not meant to be created directly by users.  Use
        the [`microgrid.new_battery_pool`][frequenz.sdk.microgrid.new_battery_pool]
        method for creating `BatteryPool` instances.

    Args:
        pool_ref_store: The battery 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 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._batteries,
            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 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._batteries,
            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 set of batteries.

Power values need to follow the Passive Sign Convention (PSC). That is, positive values indicate charge power and negative values indicate discharge power.

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

PARAMETER	DESCRIPTION
power	The power to propose for the batteries in the pool. 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, this will limit the bounds for lower priority actors. TYPE: Bounds[Power | None] DEFAULT: Bounds(None, None)

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

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

    Power values need to follow the Passive Sign Convention (PSC). That is, positive
    values indicate charge power and negative values indicate discharge power.

    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 for the batteries in the pool.  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, this 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._batteries,
            priority=self._priority,
            creation_time=asyncio.get_running_loop().time(),
        )
    )

stop async

stop() -> None

Stop all tasks and channels owned by the BatteryPool.

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

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

frequenz.sdk.timeseries.battery_pool.BatteryPoolReport

Bases: Protocol

A status report for a battery pool.

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

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

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