Remove redundant async loops and declarations

Signed-off-by: Sahas Subramanian <[email protected]>

Author: shsms

src/frequenz/client/reporting/_client.py

@@ -305,17 +305,16 @@ def transform_response(
             ) -> ComponentsDataBatch:
                 return ComponentsDataBatch(response)
 
-            async def stream_method() -> (
+            def stream_method() -> (
                 AsyncIterable[PBReceiveMicrogridComponentsDataStreamResponse]
             ):
                 call_iterator = self.stub.ReceiveMicrogridComponentsDataStream(
                     request, metadata=self._metadata
                 )
-                async for response in cast(
+                return cast(
                     AsyncIterable[PBReceiveMicrogridComponentsDataStreamResponse],
                     call_iterator,
-                ):
-                    yield response
+                )
 
             self._components_data_streams[stream_key] = GrpcStreamBroadcaster(
                 stream_name="microgrid-components-data-stream",
@@ -491,17 +490,16 @@ def transform_response(
             ) -> SensorsDataBatch:
                 return SensorsDataBatch(response)
 
-            async def stream_method() -> (
+            def stream_method() -> (
                 AsyncIterable[PBReceiveMicrogridSensorsDataStreamResponse]
             ):
                 call_iterator = self.stub.ReceiveMicrogridSensorsDataStream(
                     request, metadata=self._metadata
                 )
-                async for response in cast(
+                return cast(
                     AsyncIterable[PBReceiveMicrogridSensorsDataStreamResponse],
                     call_iterator,
-                ):
-                    yield response
+                )
 
             self._sensors_data_streams[stream_key] = GrpcStreamBroadcaster(
                 stream_name="microgrid-sensors-data-stream",
@@ -511,7 +509,7 @@ async def stream_method() -> (
 
         return self._sensors_data_streams[stream_key].new_receiver()
 
-    async def receive_aggregated_data(
+    def receive_aggregated_data(
         self,
         *,
         microgrid_id: int,
@@ -587,17 +585,14 @@ def transform_response(
             ) -> MetricSample:
                 return AggregatedMetric(response).sample()
 
-            async def stream_method() -> AsyncIterable[PBAggregatedStreamResponse]:
+            def stream_method() -> AsyncIterable[PBAggregatedStreamResponse]:
                 call_iterator = (
                     self.stub.ReceiveAggregatedMicrogridComponentsDataStream(
                         request, metadata=self._metadata
                     )
                 )
 
-                async for response in cast(
-                    AsyncIterable[PBAggregatedStreamResponse], call_iterator
-                ):
-                    yield response
+                return cast(AsyncIterable[PBAggregatedStreamResponse], call_iterator)
 
             self._aggregated_data_streams[stream_key] = GrpcStreamBroadcaster(
                 stream_name="aggregated-microgrid-data-stream",

src/frequenz/client/reporting/cli/__main__.py

@@ -171,7 +171,7 @@ async def data_iter() -> AsyncIterator[MetricSample]:
         for formula in formulas:
             assert resampling_period is not None
             for metric in metrics:
-                async for sample in await client.receive_aggregated_data(
+                async for sample in client.receive_aggregated_data(
                     microgrid_id=microgrid_id,
                     metric=metric,
                     aggregation_formula=formula,