Control interface for BatteryPool (frequenz-floss#408)

This PR updates the PowerDistributingActor to remove its multi-user
interface, make its accessible only through the BatteryPool, with these
methods:

1. set_power :: send a PSC power request to the power distributor.
2. charge :: send a charge power (+ve) to the power distributor.
3. discharge :: send a discharge power (+ve) to the power distributor.
4. power_distribution_results :: get a receiver to power distributor
responses.

Author: shsms

benchmarks/power_distribution/power_distributor.py

@@ -7,18 +7,16 @@
 import csv
 import random
 import timeit
-from dataclasses import dataclass
 from datetime import timedelta
 from typing import Any, Coroutine, Dict, List, Set  # pylint: disable=unused-import
 
-from frequenz.channels import Bidirectional, Broadcast
+from frequenz.channels import Broadcast
 
 from frequenz.sdk import microgrid
-from frequenz.sdk.actor import ResamplerConfig
+from frequenz.sdk.actor import ChannelRegistry, ResamplerConfig
 from frequenz.sdk.actor.power_distributing import (
     BatteryStatus,
     Error,
-    Ignored,
     OutOfBound,
     PartialFailure,
     PowerDistributingActor,
@@ -33,15 +31,7 @@
 PORT = 61060
 
 
-@dataclass
-class User:
-    """User definition."""
-
-    user_id: str
-    channel: Bidirectional.Handle[Request, Result]
-
-
-async def run_user(user: User, batteries: Set[int], request_num: int) -> List[Result]:
+async def send_requests(batteries: Set[int], request_num: int) -> List[Result]:
     """Send requests to the PowerDistributingActor and wait for the response.
 
     Args:
@@ -55,15 +45,15 @@ async def run_user(user: User, batteries: Set[int], request_num: int) -> List[Re
     Returns:
         List of the results from the PowerDistributingActor.
     """
+    battery_pool = microgrid.battery_pool(batteries)
+    results_rx = battery_pool.power_distribution_results()
     result: List[Result] = []
     for _ in range(request_num):
-        await user.channel.send(
-            Request(power=float(random.randrange(100000, 1000000)), batteries=batteries)
-        )
+        await battery_pool.set_power(float(random.randrange(100000, 1000000)))
         try:
-            output = await asyncio.wait_for(user.channel.receive(), timeout=3)
+            output = await asyncio.wait_for(results_rx.receive(), timeout=3)
             if output is None:
-                raise SystemError(f"Channel for {user.user_id} closed!")
+                raise SystemError(f"Power response channel for {battery_pool} closed!")
             result.append(output)
         except asyncio.exceptions.TimeoutError:
             print("TIMEOUT ERROR")
@@ -82,7 +72,6 @@ def parse_result(result: List[List[Result]]) -> Dict[str, float]:
     """
     result_counts = {
         Error: 0,
-        Ignored: 0,
         Success: 0,
         PartialFailure: 0,
         OutOfBound: 0,
@@ -95,57 +84,45 @@ def parse_result(result: List[List[Result]]) -> Dict[str, float]:
     return {
         "success_num": result_counts[Success],
         "failed_num": result_counts[PartialFailure],
-        "ignore_num": result_counts[Ignored],
         "error_num": result_counts[Error],
         "out_of_bound": result_counts[OutOfBound],
     }
 
 
 async def run_test(  # pylint: disable=too-many-locals
-    users_num: int,
-    requests_per_user: int,
+    num_requests: int,
     batteries: Set[int],
 ) -> Dict[str, Any]:
     """Run test.
 
     Args:
-        users_num: Number of users to register
-        requests_per_user: How many request user should send.
+        num_requests: Number of requests to send.
         batteries: Set of batteries for each request.
 
     Returns:
         Dictionary with statistics.
     """
     start = timeit.default_timer()
 
-    channels: Dict[str, Bidirectional[Request, Result]] = {
-        str(user_id): Bidirectional[Request, Result](str(user_id), "power_distributor")
-        for user_id in range(users_num)
-    }
-
-    service_channels = {
-        user_id: channel.service_handle for user_id, channel in channels.items()
-    }
-
+    power_request_channel = Broadcast[Request]("power-request")
     battery_status_channel = Broadcast[BatteryStatus]("battery-status")
-
+    channel_registry = ChannelRegistry(name="power_distributor")
     distributor = PowerDistributingActor(
-        service_channels, battery_status_sender=battery_status_channel.new_sender()
+        channel_registry=channel_registry,
+        requests_receiver=power_request_channel.new_receiver(),
+        battery_status_sender=battery_status_channel.new_sender(),
     )
 
     tasks: List[Coroutine[Any, Any, List[Result]]] = []
-    for user_id, channel in channels.items():
-        user = User(user_id, channel.client_handle)
-        tasks.append(run_user(user, batteries, requests_per_user))
+    tasks.append(send_requests(batteries, num_requests))
 
     result = await asyncio.gather(*tasks)
     exec_time = timeit.default_timer() - start
 
     await distributor._stop()  # type: ignore # pylint: disable=no-member, protected-access
 
     summary = parse_result(result)
-    summary["users_num"] = users_num
-    summary["requests_per_user"] = requests_per_user
+    summary["num_requests"] = num_requests
     summary["batteries_num"] = len(batteries)
     summary["exec_time"] = exec_time
     return summary
@@ -166,12 +143,11 @@ async def run() -> None:
     # Take some time to get data from components
     await asyncio.sleep(4)
     with open("/dev/stdout", "w", encoding="utf-8") as csvfile:
-        fields = await run_test(0, 0, batteries_ids)
+        fields = await run_test(0, batteries_ids)
         out = csv.DictWriter(csvfile, fields.keys())
         out.writeheader()
-        out.writerow(await run_test(1, 1, batteries_ids))
-        out.writerow(await run_test(1, 10, batteries_ids))
-        out.writerow(await run_test(10, 10, batteries_ids))
+        out.writerow(await run_test(1, batteries_ids))
+        out.writerow(await run_test(10, batteries_ids))
 
 
 async def main() -> None:

examples/power_distribution.py

@@ -14,29 +14,14 @@
 import logging
 from datetime import datetime, timedelta, timezone
 from queue import Queue
-from typing import List, Optional, Set
+from typing import List, Optional
 
-from frequenz.channels import Bidirectional, Broadcast, Receiver, Sender
+from frequenz.channels import Broadcast, Receiver, Sender
 
 from frequenz.sdk import actor, microgrid
-from frequenz.sdk.actor import (
-    ChannelRegistry,
-    ComponentMetricRequest,
-    ComponentMetricsResamplingActor,
-    DataSourcingActor,
-    ResamplerConfig,
-)
-from frequenz.sdk.actor.power_distributing import (
-    BatteryStatus,
-    PowerDistributingActor,
-    Request,
-    Result,
-    Success,
-)
-from frequenz.sdk.microgrid import connection_manager
-from frequenz.sdk.microgrid.component import Component, ComponentCategory
+from frequenz.sdk.actor import ResamplerConfig
+from frequenz.sdk.actor.power_distributing import Result, Success
 from frequenz.sdk.timeseries import Sample
-from frequenz.sdk.timeseries.logical_meter import LogicalMeter
 
 _logger = logging.getLogger(__name__)
 HOST = "microgrid.sandbox.api.frequenz.io"  # it should be the host name.
@@ -47,30 +32,26 @@
 class DecisionMakingActor:
     """Actor that receives set receives power for given batteries."""
 
-    def __init__(  # pylint: disable=too-many-arguments
+    def __init__(
         self,
         power_channel: Receiver[List[float]],
-        power_distributor_handle: Bidirectional.Handle[Request, Result],
-        batteries: Set[int],
     ) -> None:
         """Create actor instance.
 
         Args:
             power_channel: channel where actor receives requests
-            power_distributor_handle: Channel
-                for communication with power distributor
-            batteries: Batteries to charge/discharge
         """
         self._power_channel = power_channel
-        self._power_distributor_handle = power_distributor_handle
-        self._batteries = batteries
 
     async def run(self) -> None:
         """Run actor.
 
         Raises:
             RuntimeError: If any channel was closed unexpectedly
         """
+        battery_pool = microgrid.battery_pool()
+        result_rx = battery_pool.power_distribution_results()
+
         while True:
             # wait for request with blocking
             request: Optional[List[float]] = await self._power_channel.receive()
@@ -88,16 +69,11 @@ async def run(self) -> None:
                 # Discharge
                 power_to_set = -10000.0
 
-            await self._power_distributor_handle.send(
-                Request(
-                    power_to_set,
-                    batteries=self._batteries,
-                    request_timeout_sec=2.0,
-                )
-            )
+            await battery_pool.set_power(power_to_set)
             try:
                 result: Optional[Result] = await asyncio.wait_for(
-                    self._power_distributor_handle.receive(), timeout=3
+                    result_rx.receive(),
+                    timeout=3,
                 )
             except asyncio.exceptions.TimeoutError:
                 _logger.error(
@@ -164,76 +140,22 @@ async def run() -> None:
         HOST, PORT, ResamplerConfig(resampling_period=timedelta(seconds=1.0))
     )
 
-    channel_registry = ChannelRegistry(name="Microgrid Channel Registry")
-
-    data_source_request_channel = Broadcast[ComponentMetricRequest](
-        "Data Source Request Channel"
-    )
-
-    resampling_actor_request_channel = Broadcast[ComponentMetricRequest](
-        "Resampling Actor Request Channel"
-    )
-
-    _ds_actor = DataSourcingActor(
-        request_receiver=data_source_request_channel.new_receiver(),
-        registry=channel_registry,
-    )
-
-    _resampling_actor = ComponentMetricsResamplingActor(
-        channel_registry=channel_registry,
-        data_sourcing_request_sender=data_source_request_channel.new_sender(),
-        resampling_request_receiver=resampling_actor_request_channel.new_receiver(),
-        config=ResamplerConfig(resampling_period=timedelta(seconds=1.0)),
-    )
-
-    logical_meter = LogicalMeter(
-        channel_registry,
-        resampling_actor_request_channel.new_sender(),
-    )
-    sending_actor_id: str = "SendingActor"
-    # Bidirectional channel is used for one sender - one receiver communication
-    power_distributor_channels = {
-        sending_actor_id: Bidirectional[Request, Result](
-            client_id=sending_actor_id, service_id="PowerDistributingActor"
-        )
-    }
-
-    battery_status_channel = Broadcast[BatteryStatus]("battery-status")
-
-    power_distributor = PowerDistributingActor(
-        users_channels={
-            key: channel.service_handle
-            for key, channel in power_distributor_channels.items()
-        },
-        battery_status_sender=battery_status_channel.new_sender(),
-    )
-
+    logical_meter = microgrid.logical_meter()
     # Channel to communicate between actors.
     power_dist_req_chan = Broadcast[List[float]](
         "power-distribing-req", resend_latest=True
     )
 
-    # You should get components from ComponentGraph, not from the api.
-    # It is faster and and non blocking approach.
-    batteries: Set[Component] = connection_manager.get().component_graph.components(
-        # component_type=set(ComponentType.BATTERY) in v0.8.0
-        component_category={ComponentCategory.BATTERY}
-    )
-
     service_actor = DecisionMakingActor(
         power_channel=power_dist_req_chan.new_receiver(),
-        power_distributor_handle=power_distributor_channels[
-            sending_actor_id
-        ].client_handle,
-        batteries={battery.component_id for battery in batteries},
     )
 
     client_actor = DataCollectingActor(
         request_channel=power_dist_req_chan.new_sender(),
         active_power_data=logical_meter.grid_power.new_receiver(),
     )
 
-    await actor.run(service_actor, client_actor, power_distributor)
+    await actor.run(service_actor, client_actor)
 
 
 asyncio.run(run())

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

@@ -60,7 +60,10 @@ def get_channel_name(self) -> str:
         Returns:
             A string denoting a channel name.
         """
-        return f"{self.component_id}::{self.metric_id.name}::{self.start_time}::{self.namespace}"
+        return (
+            f"component-stream::{self.component_id}::{self.metric_id.name}::"
+            f"{self.start_time}::{self.namespace}"
+        )
 
 
 _MeterDataMethods: Dict[ComponentMetricId, Callable[[MeterData], float]] = {

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

@@ -13,15 +13,14 @@
 from ._battery_pool_status import BatteryStatus
 from .power_distributing import PowerDistributingActor
 from .request import Request
-from .result import Error, Ignored, OutOfBound, PartialFailure, Result, Success
+from .result import Error, OutOfBound, PartialFailure, Result, Success
 
 __all__ = [
     "PowerDistributingActor",
     "Request",
     "Result",
     "Error",
     "Success",
-    "Ignored",
     "OutOfBound",
     "PartialFailure",
     "BatteryStatus",

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

@@ -6,6 +6,7 @@
 
 import asyncio
 import logging
+from collections import abc
 from dataclasses import dataclass
 from typing import Dict, Set
 
@@ -28,7 +29,7 @@ class BatteryStatus:
     uncertain: Set[int]
     """Set of batteries that should be used only if there are no working batteries."""
 
-    def get_working_batteries(self, batteries: Set[int]) -> Set[int]:
+    def get_working_batteries(self, batteries: abc.Set[int]) -> Set[int]:
         """From the given set of batteries return working batteries.
 
         Args:
@@ -203,7 +204,7 @@ async def update_status(
             SetPowerResult(succeed_batteries, failed_batteries)
         )
 
-    def get_working_batteries(self, batteries: Set[int]) -> Set[int]:
+    def get_working_batteries(self, batteries: abc.Set[int]) -> Set[int]:
         """From the given set of batteries get working.
 
         Args: