Miscelaneous improvements and fixes (#953)

- **tests: Remove unnecessary class wrapper**
- **Update sandbox port to the newest sandbox**
- **Fix invalid `Power` constructor**
- **Log unhandled exceptions in the Microgrid API source**

Author: llucax

benchmarks/power_distribution/power_distributor.py

@@ -30,7 +30,7 @@
 from frequenz.sdk.timeseries._quantities import Power
 
 HOST = "microgrid.sandbox.api.frequenz.io"
-PORT = 61060
+PORT = 62060
 
 
 # TODO: this send_requests function uses the battery pool to # pylint: disable=fixme
@@ -55,7 +55,7 @@ async def send_requests(batteries: set[int], request_num: int) -> list[Result]:
     result: list[Any] = []
     for _ in range(request_num):
         await battery_pool.propose_power(
-            Power(float(random.randrange(100000, 1000000)))
+            Power.from_watts(float(random.randrange(100000, 1000000)))
         )
         try:
             output = await asyncio.wait_for(results_rx.receive(), timeout=3)

examples/battery_pool.py

@@ -14,7 +14,7 @@
 from frequenz.sdk.actor import ResamplerConfig
 
 HOST = "microgrid.sandbox.api.frequenz.io"  # it should be the host name.
-PORT = 61060
+PORT = 62060
 
 
 async def main() -> None:

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

@@ -383,48 +383,62 @@ async def _handle_data_stream(
         Args:
             comp_id: Id of the requested component.
             category: The category of the component.
+
+        Raises:
+            Exception: if an error occurs while handling the data stream.
         """
-        stream_senders = []
-        if comp_id in self._req_streaming_metrics:
-            await self._check_requested_component_and_metrics(
-                comp_id, category, self._req_streaming_metrics[comp_id]
+        try:
+            stream_senders = []
+            if comp_id in self._req_streaming_metrics:
+                await self._check_requested_component_and_metrics(
+                    comp_id, category, self._req_streaming_metrics[comp_id]
+                )
+                stream_senders = self._get_metric_senders(
+                    category, self._req_streaming_metrics[comp_id]
+                )
+            api_data_receiver: Receiver[Any] = self.comp_data_receivers[comp_id]
+
+            senders_done: asyncio.Event = asyncio.Event()
+            pending_messages = 0
+
+            def process_msg(data: Any) -> None:
+                tasks = []
+                for extractor, senders in stream_senders:
+                    for sender in senders:
+                        tasks.append(
+                            sender.send(
+                                Sample(data.timestamp, Quantity(extractor(data)))
+                            )
+                        )
+                asyncio.gather(*tasks)
+                nonlocal pending_messages
+                pending_messages -= 1
+                if pending_messages == 0:
+                    senders_done.set()
+
+            async for data in api_data_receiver:
+                pending_messages += 1
+                senders_done.clear()
+                process_msg(data)
+
+            while pending_messages > 0:
+                await senders_done.wait()
+
+            await asyncio.gather(
+                *[
+                    self._registry.close_and_remove(r.get_channel_name())
+                    for requests in self._req_streaming_metrics[comp_id].values()
+                    for r in requests
+                ]
             )
-            stream_senders = self._get_metric_senders(
-                category, self._req_streaming_metrics[comp_id]
+        except Exception:
+            _logger.exception(
+                "Unexpected error while handling data stream for component %d (%s), "
+                "component data is not being streamed anymore",
+                comp_id,
+                category.name,
             )
-        api_data_receiver: Receiver[Any] = self.comp_data_receivers[comp_id]
-
-        senders_done: asyncio.Event = asyncio.Event()
-        pending_messages = 0
-
-        def process_msg(data: Any) -> None:
-            tasks = []
-            for extractor, senders in stream_senders:
-                for sender in senders:
-                    tasks.append(
-                        sender.send(Sample(data.timestamp, Quantity(extractor(data))))
-                    )
-            asyncio.gather(*tasks)
-            nonlocal pending_messages
-            pending_messages -= 1
-            if pending_messages == 0:
-                senders_done.set()
-
-        async for data in api_data_receiver:
-            pending_messages += 1
-            senders_done.clear()
-            process_msg(data)
-
-        while pending_messages > 0:
-            await senders_done.wait()
-
-        await asyncio.gather(
-            *[
-                self._registry.close_and_remove(r.get_channel_name())
-                for requests in self._req_streaming_metrics[comp_id].values()
-                for r in requests
-            ]
-        )
+            raise
 
     async def _update_streams(
         self,

tests/actor/test_data_sourcing.py

@@ -19,90 +19,81 @@
 # pylint: disable=no-member
 
 
-class TestDataSourcingActor:
+async def test_data_sourcing_actor() -> None:
     """Tests for the DataSourcingActor."""
-
-    async def test_data_sourcing_actor(self) -> None:
-        """Tests for the DataSourcingActor."""
-        servicer = mock_api.MockMicrogridServicer()
-        server = mock_api.MockGrpcServer(servicer, port=57899)
-        await server.start()
-
-        servicer.add_component(
-            1, components_pb.ComponentCategory.COMPONENT_CATEGORY_GRID
+    servicer = mock_api.MockMicrogridServicer()
+    server = mock_api.MockGrpcServer(servicer, port=57899)
+    await server.start()
+
+    servicer.add_component(1, components_pb.ComponentCategory.COMPONENT_CATEGORY_GRID)
+    servicer.add_component(4, components_pb.ComponentCategory.COMPONENT_CATEGORY_METER)
+    servicer.add_component(7, components_pb.ComponentCategory.COMPONENT_CATEGORY_METER)
+    servicer.add_component(
+        8, components_pb.ComponentCategory.COMPONENT_CATEGORY_INVERTER
+    )
+    servicer.add_component(
+        9, components_pb.ComponentCategory.COMPONENT_CATEGORY_BATTERY
+    )
+
+    servicer.add_connection(1, 4)
+    servicer.add_connection(1, 7)
+    servicer.add_connection(7, 8)
+    servicer.add_connection(8, 9)
+
+    await connection_manager.initialize("[::1]", 57899)
+
+    req_chan = Broadcast[ComponentMetricRequest](name="data_sourcing_requests")
+    req_sender = req_chan.new_sender()
+
+    registry = ChannelRegistry(name="test-registry")
+
+    async with DataSourcingActor(req_chan.new_receiver(), registry):
+        active_power_request = ComponentMetricRequest(
+            "test-namespace", 4, ComponentMetricId.ACTIVE_POWER, None
         )
-        servicer.add_component(
-            4, components_pb.ComponentCategory.COMPONENT_CATEGORY_METER
+        active_power_recv = registry.get_or_create(
+            Sample[Quantity], active_power_request.get_channel_name()
+        ).new_receiver()
+        await req_sender.send(active_power_request)
+
+        reactive_power_request = ComponentMetricRequest(
+            "test-namespace", 4, ComponentMetricId.REACTIVE_POWER, None
         )
-        servicer.add_component(
-            7, components_pb.ComponentCategory.COMPONENT_CATEGORY_METER
+        _ = registry.get_or_create(
+            Sample[Quantity], reactive_power_request.get_channel_name()
+        ).new_receiver()
+        await req_sender.send(reactive_power_request)
+
+        soc_request = ComponentMetricRequest(
+            "test-namespace", 9, ComponentMetricId.SOC, None
         )
-        servicer.add_component(
-            8, components_pb.ComponentCategory.COMPONENT_CATEGORY_INVERTER
+        soc_recv = registry.get_or_create(
+            Sample[Quantity], soc_request.get_channel_name()
+        ).new_receiver()
+        await req_sender.send(soc_request)
+
+        soc2_request = ComponentMetricRequest(
+            "test-namespace", 9, ComponentMetricId.SOC, None
         )
-        servicer.add_component(
-            9, components_pb.ComponentCategory.COMPONENT_CATEGORY_BATTERY
+        soc2_recv = registry.get_or_create(
+            Sample[Quantity], soc2_request.get_channel_name()
+        ).new_receiver()
+        await req_sender.send(soc2_request)
+
+        for _ in range(3):
+            sample = await soc_recv.receive()
+            assert sample.value is not None
+            assert 9.0 == sample.value.base_value
+
+            sample = await soc2_recv.receive()
+            assert sample.value is not None
+            assert 9.0 == sample.value.base_value
+
+            sample = await active_power_recv.receive()
+            assert sample.value is not None
+            assert 100.0 == sample.value.base_value
+
+        assert await server.graceful_shutdown()
+        connection_manager._CONNECTION_MANAGER = (  # pylint: disable=protected-access
+            None
         )
-
-        servicer.add_connection(1, 4)
-        servicer.add_connection(1, 7)
-        servicer.add_connection(7, 8)
-        servicer.add_connection(8, 9)
-
-        await connection_manager.initialize("[::1]", 57899)
-
-        req_chan = Broadcast[ComponentMetricRequest](name="data_sourcing_requests")
-        req_sender = req_chan.new_sender()
-
-        registry = ChannelRegistry(name="test-registry")
-
-        async with DataSourcingActor(req_chan.new_receiver(), registry):
-            active_power_request = ComponentMetricRequest(
-                "test-namespace", 4, ComponentMetricId.ACTIVE_POWER, None
-            )
-            active_power_recv = registry.get_or_create(
-                Sample[Quantity], active_power_request.get_channel_name()
-            ).new_receiver()
-            await req_sender.send(active_power_request)
-
-            reactive_power_request = ComponentMetricRequest(
-                "test-namespace", 4, ComponentMetricId.REACTIVE_POWER, None
-            )
-            _ = registry.get_or_create(
-                Sample[Quantity], reactive_power_request.get_channel_name()
-            ).new_receiver()
-            await req_sender.send(reactive_power_request)
-
-            soc_request = ComponentMetricRequest(
-                "test-namespace", 9, ComponentMetricId.SOC, None
-            )
-            soc_recv = registry.get_or_create(
-                Sample[Quantity], soc_request.get_channel_name()
-            ).new_receiver()
-            await req_sender.send(soc_request)
-
-            soc2_request = ComponentMetricRequest(
-                "test-namespace", 9, ComponentMetricId.SOC, None
-            )
-            soc2_recv = registry.get_or_create(
-                Sample[Quantity], soc2_request.get_channel_name()
-            ).new_receiver()
-            await req_sender.send(soc2_request)
-
-            for _ in range(3):
-                sample = await soc_recv.receive()
-                assert sample.value is not None
-                assert 9.0 == sample.value.base_value
-
-                sample = await soc2_recv.receive()
-                assert sample.value is not None
-                assert 9.0 == sample.value.base_value
-
-                sample = await active_power_recv.receive()
-                assert sample.value is not None
-                assert 100.0 == sample.value.base_value
-
-            assert await server.graceful_shutdown()
-            connection_manager._CONNECTION_MANAGER = (  # pylint: disable=protected-access
-                None
-            )