Skip to content

battery_pool

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
Source code in frequenz/sdk/timeseries/battery_pool/_battery_pool.py
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
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,
        battery_pool_ref: 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.battery_pool`][frequenz.sdk.microgrid.battery_pool] method
            for creating `BatteryPool` instances.

        Args:
            battery_pool_ref: 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._battery_pool = battery_pool_ref
        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,
        *,
        request_timeout: timedelta = timedelta(seconds=5.0),
        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.

        If the same batteries are shared by multiple actors, the power manager will
        consider the priority of the actors, the bounds they set, and their preferred
        power, when calculating the target power for the batteries.

        The preferred power of lower priority actors will take precedence as long as
        they respect the bounds set by higher priority actors.  If lower priority actors
        request power values outside of the bounds set by higher priority actors, the
        target power will be the closest value to the preferred power that is within the
        bounds.

        When there are no other actors trying to use the same batteries, the actor's
        preferred power would be set as the target power, as long as it falls within the
        system power bounds for the batteries.

        The result of the request can be accessed using the receiver returned from the
        [`power_status`][frequenz.sdk.timeseries.battery_pool.BatteryPool.power_status]
        method, which also streams the bounds that an actor should comply with, based on
        its priority.

        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.  If both are `None`, it is equivalent to not having a
                proposal or withdrawing a previous one.
            request_timeout: The timeout for the request.
            bounds: The power bounds for the 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
                and experimental feature.
        """
        await self._battery_pool._power_manager_requests_sender.send(
            _power_managing.Proposal(
                source_id=self._source_id,
                preferred_power=power,
                bounds=bounds,
                component_ids=self._battery_pool._batteries,
                priority=self._priority,
                creation_time=asyncio.get_running_loop().time(),
                request_timeout=request_timeout,
            )
        )

    async def propose_charge(
        self,
        power: Power | None,
        *,
        request_timeout: timedelta = timedelta(seconds=5.0),
    ) -> 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.  The bounds for lower priority actors
        can't be specified with this method.  If that's required, use the
        `propose_power` method instead.

        The result of the request can be accessed using the receiver returned from the
        [`power_status`][frequenz.sdk.timeseries.battery_pool.BatteryPool.power_status]
        method.

        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.
            request_timeout: The timeout for the request.

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

    async def propose_discharge(
        self,
        power: Power | None,
        *,
        request_timeout: timedelta = timedelta(seconds=5.0),
    ) -> 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.  The bounds for lower priority actors
        can't be specified with this method.  If that's required, use the
        `propose_power` method instead.

        The result of the request can be accessed using the receiver returned from the
        [`power_status`][frequenz.sdk.timeseries.battery_pool.BatteryPool.power_status]
        method.

        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.
            request_timeout: The timeout for the request.

        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._battery_pool._power_manager_requests_sender.send(
            _power_managing.Proposal(
                source_id=self._source_id,
                preferred_power=power,
                bounds=timeseries.Bounds(None, None),
                component_ids=self._battery_pool._batteries,
                priority=self._priority,
                creation_time=asyncio.get_running_loop().time(),
                request_timeout=request_timeout,
            )
        )

    @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._battery_pool._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._battery_pool._formula_pool.from_power_formula_generator(
            "battery_pool_power",
            BatteryPowerFormula,
            FormulaGeneratorConfig(
                component_ids=self._battery_pool._batteries,
            ),
        )
        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._battery_pool._active_methods:
            calculator = SoCCalculator(self._battery_pool._batteries)
            self._battery_pool._active_methods[method_name] = SendOnUpdate(
                metric_calculator=calculator,
                working_batteries=self._battery_pool._working_batteries,
                min_update_interval=self._battery_pool._min_update_interval,
            )

        return self._battery_pool._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._battery_pool._active_methods:
            calculator = TemperatureCalculator(self._battery_pool._batteries)
            self._battery_pool._active_methods[method_name] = SendOnUpdate(
                metric_calculator=calculator,
                working_batteries=self._battery_pool._working_batteries,
                min_update_interval=self._battery_pool._min_update_interval,
            )
        return self._battery_pool._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._battery_pool._active_methods:
            calculator = CapacityCalculator(self._battery_pool._batteries)
            self._battery_pool._active_methods[method_name] = SendOnUpdate(
                metric_calculator=calculator,
                working_batteries=self._battery_pool._working_batteries,
                min_update_interval=self._battery_pool._min_update_interval,
            )

        return self._battery_pool._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._battery_pool._batteries,
        )
        self._battery_pool._power_bounds_subs[sub.get_channel_name()] = (
            asyncio.create_task(
                self._battery_pool._power_manager_bounds_subscription_sender.send(sub)
            )
        )
        channel = self._battery_pool._channel_registry.get_or_create(
            _power_managing._Report, sub.get_channel_name()
        )
        channel.resend_latest = True

        # More details on why the cast is needed here:
        # https://github.com/frequenz-floss/frequenz-sdk-python/issues/823
        return cast(ReceiverFetcher[BatteryPoolReport], channel)

    @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._battery_pool._active_methods:
            calculator = PowerBoundsCalculator(self._battery_pool._batteries)
            self._battery_pool._active_methods[method_name] = SendOnUpdate(
                metric_calculator=calculator,
                working_batteries=self._battery_pool._working_batteries,
                min_update_interval=self._battery_pool._min_update_interval,
            )

        return self._battery_pool._active_methods[method_name]

    async def stop(self) -> None:
        """Stop all tasks and channels owned by the BatteryPool."""
        await self._battery_pool.stop()
Attributes¤
capacity property ¤

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 ¤

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_status property ¤

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 ¤

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 ¤

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__(
    battery_pool_ref: 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.battery_pool method for creating BatteryPool instances.

PARAMETER DESCRIPTION
battery_pool_ref

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,
    battery_pool_ref: 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.battery_pool`][frequenz.sdk.microgrid.battery_pool] method
        for creating `BatteryPool` instances.

    Args:
        battery_pool_ref: 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._battery_pool = battery_pool_ref
    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,
    *,
    request_timeout: timedelta = timedelta(seconds=5.0)
) -> 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. The bounds for lower priority actors can't be specified with this method. If that's required, use the propose_power method instead.

The result of the request can be accessed using the receiver returned from the power_status method.

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

request_timeout

The timeout for the request.

TYPE: timedelta DEFAULT: timedelta(seconds=5.0)

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,
    *,
    request_timeout: timedelta = timedelta(seconds=5.0),
) -> 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.  The bounds for lower priority actors
    can't be specified with this method.  If that's required, use the
    `propose_power` method instead.

    The result of the request can be accessed using the receiver returned from the
    [`power_status`][frequenz.sdk.timeseries.battery_pool.BatteryPool.power_status]
    method.

    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.
        request_timeout: The timeout for the request.

    Raises:
        ValueError: If the given power is negative.
    """
    if power and power < Power.zero():
        raise ValueError("Charge power must be positive.")
    await self._battery_pool._power_manager_requests_sender.send(
        _power_managing.Proposal(
            source_id=self._source_id,
            preferred_power=power,
            bounds=timeseries.Bounds(None, None),
            component_ids=self._battery_pool._batteries,
            priority=self._priority,
            creation_time=asyncio.get_running_loop().time(),
            request_timeout=request_timeout,
        )
    )
propose_discharge async ¤
propose_discharge(
    power: Power | None,
    *,
    request_timeout: timedelta = timedelta(seconds=5.0)
) -> 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. The bounds for lower priority actors can't be specified with this method. If that's required, use the propose_power method instead.

The result of the request can be accessed using the receiver returned from the power_status method.

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

request_timeout

The timeout for the request.

TYPE: timedelta DEFAULT: timedelta(seconds=5.0)

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,
    *,
    request_timeout: timedelta = timedelta(seconds=5.0),
) -> 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.  The bounds for lower priority actors
    can't be specified with this method.  If that's required, use the
    `propose_power` method instead.

    The result of the request can be accessed using the receiver returned from the
    [`power_status`][frequenz.sdk.timeseries.battery_pool.BatteryPool.power_status]
    method.

    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.
        request_timeout: The timeout for the request.

    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._battery_pool._power_manager_requests_sender.send(
        _power_managing.Proposal(
            source_id=self._source_id,
            preferred_power=power,
            bounds=timeseries.Bounds(None, None),
            component_ids=self._battery_pool._batteries,
            priority=self._priority,
            creation_time=asyncio.get_running_loop().time(),
            request_timeout=request_timeout,
        )
    )
propose_power async ¤
propose_power(
    power: Power | None,
    *,
    request_timeout: timedelta = timedelta(seconds=5.0),
    bounds: 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.

If the same batteries are shared by multiple actors, the power manager will consider the priority of the actors, the bounds they set, and their preferred power, when calculating the target power for the batteries.

The preferred power of lower priority actors will take precedence as long as they respect the bounds set by higher priority actors. If lower priority actors request power values outside of the bounds set by higher priority actors, the target power will be the closest value to the preferred power that is within the bounds.

When there are no other actors trying to use the same batteries, the actor's preferred power would be set as the target power, as long as it falls within the system power bounds for the batteries.

The result of the request can be accessed using the receiver returned from the power_status method, which also streams the bounds that an actor should comply with, based on its priority.

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. If both are None, it is equivalent to not having a proposal or withdrawing a previous one.

TYPE: Power | None

request_timeout

The timeout for the request.

TYPE: timedelta DEFAULT: timedelta(seconds=5.0)

bounds

The power bounds for the 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 and experimental feature.

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,
    *,
    request_timeout: timedelta = timedelta(seconds=5.0),
    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.

    If the same batteries are shared by multiple actors, the power manager will
    consider the priority of the actors, the bounds they set, and their preferred
    power, when calculating the target power for the batteries.

    The preferred power of lower priority actors will take precedence as long as
    they respect the bounds set by higher priority actors.  If lower priority actors
    request power values outside of the bounds set by higher priority actors, the
    target power will be the closest value to the preferred power that is within the
    bounds.

    When there are no other actors trying to use the same batteries, the actor's
    preferred power would be set as the target power, as long as it falls within the
    system power bounds for the batteries.

    The result of the request can be accessed using the receiver returned from the
    [`power_status`][frequenz.sdk.timeseries.battery_pool.BatteryPool.power_status]
    method, which also streams the bounds that an actor should comply with, based on
    its priority.

    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.  If both are `None`, it is equivalent to not having a
            proposal or withdrawing a previous one.
        request_timeout: The timeout for the request.
        bounds: The power bounds for the 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
            and experimental feature.
    """
    await self._battery_pool._power_manager_requests_sender.send(
        _power_managing.Proposal(
            source_id=self._source_id,
            preferred_power=power,
            bounds=bounds,
            component_ids=self._battery_pool._batteries,
            priority=self._priority,
            creation_time=asyncio.get_running_loop().time(),
            request_timeout=request_timeout,
        )
    )
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."""
    await self._battery_pool.stop()

frequenz.sdk.timeseries.battery_pool.BatteryPoolReport ¤

Bases: Report

A status report for a battery pool.

Source code in frequenz/sdk/timeseries/battery_pool/_result_types.py
class BatteryPoolReport(Report):
    """A status report for a battery pool."""

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

    distribution_result: power_distributing.Result | None
    """The result of the last power distribution.

    This is `None` if no power distribution has been performed yet.
    """

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

        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.

distribution_result instance-attribute ¤
distribution_result: Result | None

The result of the last power distribution.

This is None if no power distribution has been performed yet.

target_power instance-attribute ¤
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.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")
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/_result_types.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.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")
        ```

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