Use new API v0.18 methods

For non data-streaming methods, update to use the new client method
names. For data-streaming methods, use the new adaptation functions in
`_old_component_data.py`.

Because `set_component_power_active()` can now return a the time when
the command expires, we also need to update the tests we use to run
them.

Note that the data-streaming method in the new client version is not
async (it returns a Receiver synchronously) so `await`s are also
removed.

For the `ComponentMetricFetcher` we also remove the `async` for the
wrapper `_subscribe()` methods.

The mock microgrid client was mostly rewritten to adjust it to the new
unified data streaming method.

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

src/frequenz/sdk/microgrid/_data_sourcing/microgrid_api_source.py

@@ -11,11 +11,7 @@
 from frequenz.channels import Receiver, Sender
 from frequenz.client.common.microgrid.components import ComponentId
 from frequenz.client.microgrid import (
-    BatteryData,
     ComponentCategory,
-    EVChargerData,
-    InverterData,
-    MeterData,
 )
 from frequenz.client.microgrid.metrics import Metric
 from frequenz.quantities import Quantity
@@ -25,6 +21,10 @@
 from ...microgrid import connection_manager
 from ...timeseries import Sample
 from .._old_component_data import (
+    BatteryData,
+    EVChargerData,
+    InverterData,
+    MeterData,
     TransitionalMetric,
 )
 from ._component_metric_request import ComponentMetricRequest
@@ -169,7 +169,7 @@ async def _get_component_category(
             return self._comp_categories_cache[comp_id]
 
         api = connection_manager.get().api_client
-        for comp in await api.components():
+        for comp in await api.list_components():
             self._comp_categories_cache[comp.component_id] = comp.category
 
         if comp_id in self._comp_categories_cache:
@@ -198,8 +198,8 @@ async def _check_battery_request(
                 _logger.error(err)
                 raise ValueError(err)
         if comp_id not in self.comp_data_receivers:
-            self.comp_data_receivers[comp_id] = (
-                await connection_manager.get().api_client.battery_data(comp_id)
+            self.comp_data_receivers[comp_id] = BatteryData.subscribe(
+                connection_manager.get().api_client, comp_id
             )
 
     async def _check_ev_charger_request(
@@ -223,8 +223,8 @@ async def _check_ev_charger_request(
                 _logger.error(err)
                 raise ValueError(err)
         if comp_id not in self.comp_data_receivers:
-            self.comp_data_receivers[comp_id] = (
-                await connection_manager.get().api_client.ev_charger_data(comp_id)
+            self.comp_data_receivers[comp_id] = EVChargerData.subscribe(
+                connection_manager.get().api_client, comp_id
             )
 
     async def _check_inverter_request(
@@ -248,8 +248,8 @@ async def _check_inverter_request(
                 _logger.error(err)
                 raise ValueError(err)
         if comp_id not in self.comp_data_receivers:
-            self.comp_data_receivers[comp_id] = (
-                await connection_manager.get().api_client.inverter_data(comp_id)
+            self.comp_data_receivers[comp_id] = InverterData.subscribe(
+                connection_manager.get().api_client, comp_id
             )
 
     async def _check_meter_request(
@@ -273,8 +273,8 @@ async def _check_meter_request(
                 _logger.error(err)
                 raise ValueError(err)
         if comp_id not in self.comp_data_receivers:
-            self.comp_data_receivers[comp_id] = (
-                await connection_manager.get().api_client.meter_data(comp_id)
+            self.comp_data_receivers[comp_id] = MeterData.subscribe(
+                connection_manager.get().api_client, comp_id
             )
 
     async def _check_requested_component_and_metrics(

src/frequenz/sdk/microgrid/_power_distributing/_component_managers/_battery_manager.py

@@ -8,21 +8,20 @@
 import logging
 import math
 import typing
-from datetime import timedelta
+from datetime import datetime, timedelta
 
 from frequenz.channels import LatestValueCache, Receiver, Sender
 from frequenz.client.common.microgrid.components import ComponentId
 from frequenz.client.microgrid import (
     ApiClientError,
-    BatteryData,
     ComponentCategory,
-    InverterData,
     OperationOutOfRange,
 )
 from frequenz.quantities import Power
 from typing_extensions import override
 
 from ... import connection_manager
+from ..._old_component_data import BatteryData, InverterData
 from .._component_pool_status_tracker import ComponentPoolStatusTracker
 from .._component_status import BatteryStatusTracker, ComponentPoolStatus
 from .._distribution_algorithm import (
@@ -313,14 +312,16 @@ async def _create_channels(self) -> None:
         api = connection_manager.get().api_client
         manager_id = f"{type(self).__name__}«{hex(id(self))}»"
         for battery_id, inverter_ids in self._bat_invs_map.items():
-            bat_recv: Receiver[BatteryData] = await api.battery_data(battery_id)
+            bat_recv: Receiver[BatteryData] = BatteryData.subscribe(api, battery_id)
             self._battery_caches[battery_id] = LatestValueCache(
                 bat_recv,
                 unique_id=f"{manager_id}:battery«{battery_id}»",
             )
 
             for inverter_id in inverter_ids:
-                inv_recv: Receiver[InverterData] = await api.inverter_data(inverter_id)
+                inv_recv: Receiver[InverterData] = InverterData.subscribe(
+                    api, inverter_id
+                )
                 self._inverter_caches[inverter_id] = LatestValueCache(
                     inv_recv, unique_id=f"{manager_id}:inverter«{inverter_id}»"
                 )
@@ -638,7 +639,7 @@ async def _set_distributed_power(
 
         tasks = {
             inverter_id: asyncio.create_task(
-                api.set_power(inverter_id, power.as_watts())
+                api.set_component_power_active(inverter_id, power.as_watts())
             )
             for inverter_id, power in distribution.distribution.items()
             if power != Power.zero()
@@ -681,7 +682,7 @@ async def _set_distributed_power(
 
     def _parse_result(
         self,
-        tasks: dict[ComponentId, asyncio.Task[None]],
+        tasks: dict[ComponentId, asyncio.Task[datetime | None]],
         distribution: dict[ComponentId, Power],
         request_timeout: timedelta,
     ) -> tuple[Power, set[ComponentId]]:

src/frequenz/sdk/microgrid/_power_distributing/_component_managers/_ev_charger_manager/_ev_charger_manager.py

@@ -20,7 +20,6 @@
 from frequenz.client.microgrid import (
     ApiClientError,
     ComponentCategory,
-    EVChargerData,
     MicrogridApiClient,
 )
 from frequenz.quantities import Power, Voltage
@@ -30,6 +29,7 @@
 from ....._internal._math import is_close_to_zero
 from .....timeseries import Sample3Phase
 from .... import _data_pipeline, connection_manager
+from ...._old_component_data import EVChargerData
 from ..._component_pool_status_tracker import ComponentPoolStatusTracker
 from ..._component_status import ComponentPoolStatus, EVChargerStatusTracker
 from ...request import Request
@@ -224,7 +224,7 @@ async def _run(self) -> None:  # pylint: disable=too-many-locals
         """Run the main event loop of the EV charger manager."""
         api = connection_manager.get().api_client
         ev_charger_data_rx = merge(
-            *[await api.ev_charger_data(evc_id) for evc_id in self._ev_charger_ids]
+            *(EVChargerData.subscribe(api, evc_id) for evc_id in self._ev_charger_ids)
         )
         target_power_rx = self._target_power_channel.new_receiver()
         latest_target_powers: dict[ComponentId, Power] = {}
@@ -309,10 +309,10 @@ async def _set_api_power(
             Power distribution result, corresponding to the result of the API
                 request.
         """
-        tasks: dict[ComponentId, asyncio.Task[None]] = {}
+        tasks: dict[ComponentId, asyncio.Task[datetime | None]] = {}
         for component_id, power in target_power_changes.items():
             tasks[component_id] = asyncio.create_task(
-                api.set_power(component_id, power.as_watts())
+                api.set_component_power_active(component_id, power.as_watts())
             )
         _, pending = await asyncio.wait(
             tasks.values(),

src/frequenz/sdk/microgrid/_power_distributing/_component_managers/_ev_charger_manager/_states.py

@@ -9,9 +9,10 @@
 from typing import Iterable
 
 from frequenz.client.common.microgrid.components import ComponentId
-from frequenz.client.microgrid import EVChargerData
 from frequenz.quantities import Power
 
+from ...._old_component_data import EVChargerData
+
 
 @dataclass
 class EvcState:

src/frequenz/sdk/microgrid/_power_distributing/_component_managers/_pv_inverter_manager/_pv_inverter_manager.py

@@ -6,21 +6,21 @@
 import asyncio
 import collections.abc
 import logging
-from datetime import timedelta
+from datetime import datetime, timedelta
 
 from frequenz.channels import LatestValueCache, Sender
 from frequenz.client.common.microgrid.components import ComponentId
 from frequenz.client.microgrid import (
     ApiClientError,
     ComponentCategory,
-    InverterData,
     InverterType,
 )
 from frequenz.quantities import Power
 from typing_extensions import override
 
 from ....._internal._math import is_close_to_zero
 from .... import connection_manager
+from ...._old_component_data import InverterData
 from ..._component_pool_status_tracker import ComponentPoolStatusTracker
 from ..._component_status import ComponentPoolStatus, PVInverterStatusTracker
 from ...request import Request
@@ -79,7 +79,7 @@ async def start(self) -> None:
         """Start the PV inverter manager."""
         self._component_data_caches = {
             inv_id: LatestValueCache(
-                await connection_manager.get().api_client.inverter_data(inv_id),
+                InverterData.subscribe(connection_manager.get().api_client, inv_id),
                 unique_id=f"{type(self).__name__}«{hex(id(self))}»:inverter«{inv_id}»",
             )
             for inv_id in self._pv_inverter_ids
@@ -188,10 +188,10 @@ async def _set_api_power(  # pylint: disable=too-many-locals
         remaining_power: Power,
     ) -> None:
         api_client = connection_manager.get().api_client
-        tasks: dict[ComponentId, asyncio.Task[None]] = {}
+        tasks: dict[ComponentId, asyncio.Task[datetime | None]] = {}
         for component_id, power in allocations.items():
             tasks[component_id] = asyncio.create_task(
-                api_client.set_power(component_id, power.as_watts())
+                api_client.set_component_power_active(component_id, power.as_watts())
             )
         _, pending = await asyncio.wait(
             tasks.values(),

src/frequenz/sdk/microgrid/_power_distributing/_component_status/_battery_status_tracker.py

@@ -26,20 +26,18 @@
 from frequenz.client.common.microgrid.components import ComponentId
 from frequenz.client.microgrid import (
     BatteryComponentState,
-    BatteryData,
     BatteryRelayState,
     ComponentCategory,
-    ComponentData,
     ErrorLevel,
     InverterComponentState,
-    InverterData,
 )
 from typing_extensions import override
 
 from frequenz.sdk._internal._asyncio import run_forever
 
 from ....actor._background_service import BackgroundService
 from ... import connection_manager
+from ..._old_component_data import BatteryData, ComponentData, InverterData
 from ._blocking_status import BlockingStatus
 from ._component_status import (
     ComponentStatus,
@@ -255,8 +253,10 @@ async def _run(
         """
         api_client = connection_manager.get().api_client
 
-        battery_receiver = await api_client.battery_data(self._battery.component_id)
-        inverter_receiver = await api_client.inverter_data(self._inverter.component_id)
+        battery_receiver = BatteryData.subscribe(api_client, self._battery.component_id)
+        inverter_receiver = InverterData.subscribe(
+            api_client, self._inverter.component_id
+        )
 
         battery = battery_receiver
         battery_timer = self._battery.data_recv_timer

src/frequenz/sdk/microgrid/_power_distributing/_component_status/_ev_charger_status_tracker.py

@@ -14,13 +14,13 @@
 from frequenz.client.microgrid import (
     EVChargerCableState,
     EVChargerComponentState,
-    EVChargerData,
 )
 from typing_extensions import override
 
 from ...._internal._asyncio import run_forever
 from ....actor._background_service import BackgroundService
 from ... import connection_manager
+from ..._old_component_data import EVChargerData
 from ._blocking_status import BlockingStatus
 from ._component_status import (
     ComponentStatus,
@@ -149,8 +149,9 @@ def _handle_set_power_result(
 
     async def _run(self) -> None:
         """Run the status tracker."""
-        api_client = connection_manager.get().api_client
-        ev_data_rx = await api_client.ev_charger_data(self._component_id)
+        ev_data_rx = EVChargerData.subscribe(
+            connection_manager.get().api_client, self._component_id
+        )
         set_power_result_rx = self._set_power_result_receiver
         missing_data_timer = Timer(self._max_data_age, SkipMissedAndDrift())
 

src/frequenz/sdk/microgrid/_power_distributing/_component_status/_pv_inverter_status_tracker.py

@@ -10,12 +10,13 @@
 from frequenz.channels import Receiver, Sender, select, selected_from
 from frequenz.channels.timer import SkipMissedAndDrift, Timer
 from frequenz.client.common.microgrid.components import ComponentId
-from frequenz.client.microgrid import InverterComponentState, InverterData
+from frequenz.client.microgrid import ComponentStateCode
 from typing_extensions import override
 
 from ...._internal._asyncio import run_forever
 from ....actor._background_service import BackgroundService
 from ... import connection_manager
+from ..._old_component_data import InverterData
 from ._blocking_status import BlockingStatus
 from ._component_status import (
     ComponentStatus,
@@ -141,8 +142,9 @@ def _handle_pv_inverter_data(self, pv_data: InverterData) -> ComponentStatusEnum
 
     async def _run(self) -> None:
         """Run the status tracker."""
-        api_client = connection_manager.get().api_client
-        pv_data_rx = await api_client.inverter_data(self._component_id)
+        pv_data_rx = InverterData.subscribe(
+            connection_manager.get().api_client, self._component_id
+        )
         set_power_result_rx = self._set_power_result_receiver
         missing_data_timer = Timer(self._max_data_age, SkipMissedAndDrift())
 

src/frequenz/sdk/microgrid/_power_distributing/_distribution_algorithm/_battery_distribution_algorithm.py

@@ -9,10 +9,10 @@
 from typing import NamedTuple, Sequence
 
 from frequenz.client.common.microgrid.components import ComponentId
-from frequenz.client.microgrid import BatteryData, InverterData
 from frequenz.quantities import Power
 
 from ...._internal._math import is_close_to_zero
+from ..._old_component_data import BatteryData, InverterData
 from ..result import PowerBounds
 
 _logger = logging.getLogger(__name__)

src/frequenz/sdk/microgrid/_power_distributing/power_distributing.py

@@ -12,7 +12,7 @@
 
 import asyncio
 import logging
-from datetime import timedelta
+from datetime import datetime, timedelta
 
 from frequenz.channels import Receiver, Sender
 from frequenz.client.common.microgrid.components import ComponentId
@@ -101,7 +101,9 @@ def __init__(  # pylint: disable=too-many-arguments
         self._result_sender = results_sender
         self._api_power_request_timeout = api_power_request_timeout
 
-        self._processing_tasks: dict[frozenset[ComponentId], asyncio.Task[None]] = {}
+        self._processing_tasks: dict[
+            frozenset[ComponentId], asyncio.Task[datetime | None]
+        ] = {}
         """Track the power request tasks currently being processed."""
 
         self._pending_requests: dict[frozenset[ComponentId], Request] = {}
@@ -176,7 +178,7 @@ def _handle_task_completion(
         self,
         req_id: frozenset[ComponentId],
         request: Request,
-        task: asyncio.Task[None],
+        task: asyncio.Task[datetime | None],
     ) -> None:
         """Handle the completion of a power request task.