Remove request_timeout parameter from propose_power methods (#1000)

Before the power manager was introduced, the `request_timeout` parameter
was used to specify a different timeout for the `set_power` calls made
to the api service.

The `propose_power` method is much more high level and it no longer
makes sense to provide a `request_timeout` parameter, because it could
conflict with other proposals and it would be hard to figure out which
timeout to use.

From now on, it may be specified at application startup, through the new
optional `api_power_request_timeout` parameter in the
`microgrid.initialize()` method.

Author: shsms

RELEASE_NOTES.md

@@ -17,11 +17,11 @@
     * `grid.current` -> `grid.current_per_phase`
     * `ev_charger_pool.current` -> `ev_charger_pool.current_per_phase`
 
-## New Features
+* Passing a `request_timeout` in calls to `*_pool.propose_power` is no longer supported.  It may be specified at application startup, through the new optional `api_power_request_timeout` parameter in the `microgrid.initialize()` method.
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+## New Features
 
-- Classes Bounds and SystemBounds now work with the `in` operator
+- Classes `Bounds` and `SystemBounds` now implement the `__contains__` method, allowing the use of the `in` operator to check whether a value falls within the bounds or not.
 
 ## Bug Fixes
 

benchmarks/power_distribution/power_distributor.py

@@ -120,6 +120,7 @@ async def run_test(  # pylint: disable=too-many-locals
         requests_receiver=power_request_channel.new_receiver(),
         results_sender=power_result_channel.new_sender(),
         component_pool_status_sender=battery_status_channel.new_sender(),
+        api_power_request_timeout=timedelta(seconds=5.0),
     ):
         tasks: list[Coroutine[Any, Any, list[Result]]] = []
         tasks.append(send_requests(batteries, num_requests))

src/frequenz/sdk/actor/_power_managing/_base_classes.py

@@ -7,7 +7,6 @@
 
 import abc
 import dataclasses
-import datetime
 import enum
 import typing
 
@@ -164,9 +163,6 @@ class Proposal:
     This is used by the power manager to determine the age of the proposal.
     """
 
-    request_timeout: datetime.timedelta = datetime.timedelta(seconds=5.0)
-    """The maximum amount of time to wait for the request to be fulfilled."""
-
     set_operating_point: bool
     """Whether this proposal sets the operating point power or the normal power."""
 

src/frequenz/sdk/actor/_power_managing/_power_managing_actor.py

@@ -330,15 +330,11 @@ async def _send_updated_target_power(
             proposal,
             must_send,
         )
-        request_timeout = (
-            proposal.request_timeout if proposal else timedelta(seconds=5.0)
-        )
         if target_power is not None:
             await self._power_distributing_requests_sender.send(
                 power_distributing.Request(
                     power=target_power,
                     component_ids=component_ids,
-                    request_timeout=request_timeout,
                     adjust_power=True,
                 )
             )

src/frequenz/sdk/actor/power_distributing/_component_managers/_battery_manager.py

@@ -125,14 +125,15 @@ def _add(key: str, value: dict[int, set[int]] | None) -> None:
     return mapping
 
 
-class BatteryManager(ComponentManager):
+class BatteryManager(ComponentManager):  # pylint: disable=too-many-instance-attributes
     """Class to manage the data streams for batteries."""
 
     @override
     def __init__(
         self,
         component_pool_status_sender: Sender[ComponentPoolStatus],
         results_sender: Sender[Result],
+        api_power_request_timeout: timedelta,
     ):
         """Initialize this instance.
 
@@ -142,8 +143,11 @@ def __init__(
                 streams, to dynamically adjust the values based on the health of the
                 individual batteries.
             results_sender: Channel sender to send the power distribution results to.
+            api_power_request_timeout: Timeout to use when making power requests to
+                the microgrid API.
         """
         self._results_sender = results_sender
+        self._api_power_request_timeout = api_power_request_timeout
         self._batteries = connection_manager.get().component_graph.components(
             component_categories={ComponentCategory.BATTERY}
         )
@@ -277,7 +281,7 @@ async def _distribute_power(
         )
 
         failed_power, failed_batteries = await self._set_distributed_power(
-            distribution, request.request_timeout
+            distribution, self._api_power_request_timeout
         )
 
         response: Success | PartialFailure

src/frequenz/sdk/actor/power_distributing/_component_managers/_ev_charger_manager/_ev_charger_manager.py

@@ -32,8 +32,6 @@
 
 _logger = logging.getLogger(__name__)
 
-_DEFAULT_API_REQUEST_TIMEOUT = timedelta(seconds=5.0)
-
 
 class EVChargerManager(ComponentManager):
     """Manage ev chargers for the power distributor."""
@@ -43,15 +41,19 @@ def __init__(
         self,
         component_pool_status_sender: Sender[ComponentPoolStatus],
         results_sender: Sender[Result],
+        api_power_request_timeout: timedelta,
     ):
         """Initialize the ev charger data manager.
 
         Args:
             component_pool_status_sender: Channel for sending information about which
                 components are expected to be working.
             results_sender: Channel for sending results of power distribution.
+            api_power_request_timeout: Timeout to use when making power requests to
+                the microgrid API.
         """
         self._results_sender = results_sender
+        self._api_power_request_timeout = api_power_request_timeout
         self._ev_charger_ids = self._get_ev_charger_ids()
         self._evc_states = EvcStates()
         self._voltage_cache: LatestValueCache[Sample3Phase[Voltage]] = LatestValueCache(
@@ -226,7 +228,6 @@ async def _run(self) -> None:  # pylint: disable=too-many-locals
             *[await api.ev_charger_data(evc_id) for evc_id in self._ev_charger_ids]
         )
         target_power_rx = self._target_power_channel.new_receiver()
-        api_request_timeout = _DEFAULT_API_REQUEST_TIMEOUT
         latest_target_powers: dict[int, Power] = {}
         async for selected in select(ev_charger_data_rx, target_power_rx):
             target_power_changes = {}
@@ -260,7 +261,6 @@ async def _run(self) -> None:  # pylint: disable=too-many-locals
 
             elif selected_from(selected, target_power_rx):
                 self._latest_request = selected.message
-                api_request_timeout = selected.message.request_timeout
                 self._target_power = selected.message.power
                 _logger.debug("New target power: %s", self._target_power)
                 used_power = self._evc_states.get_ev_total_used_power()
@@ -288,7 +288,7 @@ async def _run(self) -> None:  # pylint: disable=too-many-locals
 
             latest_target_powers.update(target_power_changes)
             result = await self._set_api_power(
-                api, target_power_changes, api_request_timeout
+                api, target_power_changes, self._api_power_request_timeout
             )
             await self._results_sender.send(result)
 

src/frequenz/sdk/actor/power_distributing/_component_managers/_pv_inverter_manager/_pv_inverter_manager.py

@@ -38,15 +38,19 @@ def __init__(
         self,
         component_pool_status_sender: Sender[ComponentPoolStatus],
         results_sender: Sender[Result],
+        api_power_request_timeout: timedelta,
     ) -> None:
         """Initialize this instance.
 
         Args:
             component_pool_status_sender: Channel for sending information about which
                 components are expected to be working.
             results_sender: Channel for sending results of power distribution.
+            api_power_request_timeout: Timeout to use when making power requests to
+                the microgrid API.
         """
         self._results_sender = results_sender
+        self._api_power_request_timeout = api_power_request_timeout
         self._pv_inverter_ids = self._get_pv_inverter_ids()
 
         self._component_pool_status_tracker = (
@@ -177,7 +181,7 @@ async def _set_api_power(  # pylint: disable=too-many-locals
             )
         _, pending = await asyncio.wait(
             tasks.values(),
-            timeout=request.request_timeout.total_seconds(),
+            timeout=self._api_power_request_timeout.total_seconds(),
             return_when=asyncio.ALL_COMPLETED,
         )
         # collect the timed out tasks and cancel them while keeping the

src/frequenz/sdk/actor/power_distributing/power_distributing.py

@@ -12,6 +12,8 @@
 """
 
 
+from datetime import timedelta
+
 from frequenz.channels import Receiver, Sender
 from frequenz.client.microgrid import ComponentCategory, ComponentType, InverterType
 from typing_extensions import override
@@ -60,6 +62,7 @@ def __init__(  # pylint: disable=too-many-arguments
         results_sender: Sender[Result],
         component_pool_status_sender: Sender[ComponentPoolStatus],
         *,
+        api_power_request_timeout: timedelta,
         component_category: ComponentCategory,
         component_type: ComponentType | None = None,
         name: str | None = None,
@@ -72,6 +75,8 @@ def __init__(  # pylint: disable=too-many-arguments
             results_sender: Sender for sending results to the power manager.
             component_pool_status_sender: Channel for sending information about which
                 components are expected to be working.
+            api_power_request_timeout: Timeout to use when making power requests to
+                the microgrid API.
             component_category: The category of the components that this actor is
                 responsible for.
             component_type: The type of the component of the given category that this
@@ -92,22 +97,23 @@ def __init__(  # pylint: disable=too-many-arguments
         self._component_type = component_type
         self._requests_receiver = requests_receiver
         self._result_sender = results_sender
+        self._api_power_request_timeout = api_power_request_timeout
 
         self._component_manager: ComponentManager
         if component_category == ComponentCategory.BATTERY:
             self._component_manager = BatteryManager(
-                component_pool_status_sender, results_sender
+                component_pool_status_sender, results_sender, api_power_request_timeout
             )
         elif component_category == ComponentCategory.EV_CHARGER:
             self._component_manager = EVChargerManager(
-                component_pool_status_sender, results_sender
+                component_pool_status_sender, results_sender, api_power_request_timeout
             )
         elif (
             component_category == ComponentCategory.INVERTER
             and component_type == InverterType.SOLAR
         ):
             self._component_manager = PVManager(
-                component_pool_status_sender, results_sender
+                component_pool_status_sender, results_sender, api_power_request_timeout
             )
         else:
             raise ValueError(

src/frequenz/sdk/actor/power_distributing/request.py

@@ -5,7 +5,6 @@
 
 import dataclasses
 from collections import abc
-from datetime import timedelta
 
 from ...timeseries._quantities import Power
 
@@ -20,9 +19,6 @@ class Request:
     component_ids: abc.Set[int]
     """The component ids of the components to be used for this request."""
 
-    request_timeout: timedelta = timedelta(seconds=5.0)
-    """The maximum amount of time to wait for the request to be fulfilled."""
-
     adjust_power: bool = True
     """Whether to adjust the power to match the bounds.
 

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -14,6 +14,7 @@
 import typing
 from collections import abc
 from dataclasses import dataclass
+from datetime import timedelta
 
 from frequenz.channels import Broadcast, Sender
 from frequenz.client.microgrid import ComponentCategory, InverterType
@@ -79,11 +80,14 @@ class _DataPipeline:  # pylint: disable=too-many-instance-attributes
     def __init__(
         self,
         resampler_config: ResamplerConfig,
+        api_power_request_timeout: timedelta = timedelta(seconds=5.0),
     ) -> None:
         """Create a `DataPipeline` instance.
 
         Args:
             resampler_config: Config to pass on to the resampler.
+            api_power_request_timeout: Timeout to use when making power requests to
+                the microgrid API.
         """
         from ..actor import ChannelRegistry
 
@@ -97,13 +101,18 @@ def __init__(
         self._resampling_actor: _ActorInfo | None = None
 
         self._battery_power_wrapper = PowerWrapper(
-            self._channel_registry, component_category=ComponentCategory.BATTERY
+            self._channel_registry,
+            api_power_request_timeout=api_power_request_timeout,
+            component_category=ComponentCategory.BATTERY,
         )
         self._ev_power_wrapper = PowerWrapper(
-            self._channel_registry, component_category=ComponentCategory.EV_CHARGER
+            self._channel_registry,
+            api_power_request_timeout=api_power_request_timeout,
+            component_category=ComponentCategory.EV_CHARGER,
         )
         self._pv_power_wrapper = PowerWrapper(
             self._channel_registry,
+            api_power_request_timeout=api_power_request_timeout,
             component_category=ComponentCategory.INVERTER,
             component_type=InverterType.SOLAR,
         )
@@ -485,11 +494,18 @@ async def _stop(self) -> None:
 _DATA_PIPELINE: _DataPipeline | None = None
 
 
-async def initialize(resampler_config: ResamplerConfig) -> None:
+async def initialize(
+    resampler_config: ResamplerConfig,
+    api_power_request_timeout: timedelta = timedelta(seconds=5.0),
+) -> None:
     """Initialize a `DataPipeline` instance.
 
     Args:
         resampler_config: Config to pass on to the resampler.
+        api_power_request_timeout: Timeout to use when making power requests to
+            the microgrid API.  When requests to components timeout, they will
+            be marked as blocked for a short duration, during which time they
+            will be unavailable from the corresponding component pools.
 
     Raises:
         RuntimeError: if the DataPipeline is already initialized.
@@ -498,7 +514,7 @@ async def initialize(resampler_config: ResamplerConfig) -> None:
 
     if _DATA_PIPELINE is not None:
         raise RuntimeError("DataPipeline is already initialized.")
-    _DATA_PIPELINE = _DataPipeline(resampler_config)
+    _DATA_PIPELINE = _DataPipeline(resampler_config, api_power_request_timeout)
 
 
 def frequency() -> GridFrequency: