Add Quantity types and update formulas to produce typed Samples (#422)

Author: shsms

benchmarks/power_distribution/power_distributor.py

@@ -26,6 +26,7 @@
 )
 from frequenz.sdk.microgrid import connection_manager
 from frequenz.sdk.microgrid.component import Component, ComponentCategory
+from frequenz.sdk.timeseries._quantities import Power
 
 HOST = "microgrid.sandbox.api.frequenz.io"
 PORT = 61060
@@ -49,7 +50,7 @@ async def send_requests(batteries: Set[int], request_num: int) -> List[Result]:
     results_rx = battery_pool.power_distribution_results()
     result: List[Result] = []
     for _ in range(request_num):
-        await battery_pool.set_power(float(random.randrange(100000, 1000000)))
+        await battery_pool.set_power(Power(float(random.randrange(100000, 1000000))))
         try:
             output = await asyncio.wait_for(results_rx.receive(), timeout=3)
             if output is None:

benchmarks/timeseries/periodic_feature_extractor.py

@@ -24,17 +24,18 @@
 from numpy.typing import NDArray
 
 from frequenz.sdk.timeseries import MovingWindow, PeriodicFeatureExtractor, Sample
+from frequenz.sdk.timeseries._quantities import Quantity
 
 
 async def init_feature_extractor(period: int) -> PeriodicFeatureExtractor:
     """Initialize the PeriodicFeatureExtractor class."""
     # We only need the moving window to initialize the PeriodicFeatureExtractor class.
-    lm_chan = Broadcast[Sample]("lm_net_power")
+    lm_chan = Broadcast[Sample[Quantity]]("lm_net_power")
     moving_window = MovingWindow(
         timedelta(seconds=1), lm_chan.new_receiver(), timedelta(seconds=1)
     )
 
-    await lm_chan.new_sender().send(Sample(datetime.now(tz=timezone.utc), 0))
+    await lm_chan.new_sender().send(Sample(datetime.now(tz=timezone.utc), Quantity(0)))
 
     # Initialize the PeriodicFeatureExtractor class with a period of period seconds.
     # This works since the sampling period is set to 1 second.

benchmarks/timeseries/resampling.py

@@ -8,6 +8,7 @@
 from typing import Sequence
 
 from frequenz.sdk.timeseries import Sample
+from frequenz.sdk.timeseries._quantities import Quantity
 from frequenz.sdk.timeseries._resampling import (
     ResamplerConfig,
     SourceProperties,
@@ -16,7 +17,7 @@
 
 
 def nop(  # pylint: disable=unused-argument
-    samples: Sequence[Sample],
+    samples: Sequence[Sample[Quantity]],
     resampler_config: ResamplerConfig,
     source_properties: SourceProperties,
 ) -> float:
@@ -44,7 +45,7 @@ def _do_work() -> None:
         for _n_resample in range(resamples):
             for _n_sample in range(samples):
                 now = now + timedelta(seconds=1 / samples)
-                helper.add_sample(Sample(now, 0.0))
+                helper.add_sample(Sample(now, Quantity(0.0)))
             helper.resample(now)
 
     print(timeit(_do_work, number=5))

benchmarks/timeseries/ringbuffer_memusage.py

@@ -12,6 +12,7 @@
 import numpy as np
 
 from frequenz.sdk.timeseries import Sample
+from frequenz.sdk.timeseries._quantities import Quantity
 from frequenz.sdk.timeseries._ringbuffer import OrderedRingBuffer
 
 FIVE_MINUTES = timedelta(minutes=5)
@@ -75,7 +76,7 @@ def main(ringbuffer_len: int, iterations: int, gap_size: int) -> None:
                 datetime.fromtimestamp(
                     200 + i * FIVE_MINUTES.total_seconds(), tz=timezone.utc
                 ),
-                i,
+                Quantity(i),
             )
         )
 

benchmarks/timeseries/ringbuffer_serialization.py

@@ -15,6 +15,7 @@
 
 import frequenz.sdk.timeseries._ringbuffer as rb
 from frequenz.sdk.timeseries import Sample
+from frequenz.sdk.timeseries._quantities import Quantity
 
 FILE_NAME = "ringbuffer.pkl"
 FIVE_MINUTES = timedelta(minutes=5)
@@ -72,7 +73,7 @@ def main() -> None:
                 datetime.fromtimestamp(
                     200 + i * FIVE_MINUTES.total_seconds(), tz=timezone.utc
                 ),
-                i,
+                Quantity(i),
             )
         )
 

examples/power_distribution.py

@@ -22,6 +22,7 @@
 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._quantities import Power
 
 _logger = logging.getLogger(__name__)
 HOST = "microgrid.sandbox.api.frequenz.io"  # it should be the host name.
@@ -61,13 +62,13 @@ async def run(self) -> None:
 
             avg_power = sum(request) / len(request)
             _logger.debug("Avg power %d", avg_power)
-            power_to_set: float
+            power_to_set: Power
             if avg_power > 30000:
                 # Charge
-                power_to_set = 10000.0
+                power_to_set = Power(10000.0)
             else:
                 # Discharge
-                power_to_set = -10000.0
+                power_to_set = Power(-10000.0)
 
             await battery_pool.set_power(power_to_set)
             try:
@@ -97,7 +98,7 @@ class DataCollectingActor:
     def __init__(
         self,
         request_channel: Sender[List[float]],
-        active_power_data: Receiver[Sample],
+        active_power_data: Receiver[Sample[Power]],
     ) -> None:
         """Create actor instance.
 
@@ -126,7 +127,9 @@ async def run(self) -> None:
                 if (datetime.now(timezone.utc) - time_data).total_seconds() > 30:
                     _logger.error("Active power data are stale")
                     continue
-                queue.put_nowait(active_power.value)
+                queue.put_nowait(
+                    None if not active_power.value else active_power.value.base_value
+                )
 
             await self._request_channel.send(list(queue.queue))
 

examples/resampling.py

@@ -19,6 +19,7 @@
 from frequenz.sdk.microgrid import connection_manager
 from frequenz.sdk.microgrid.component import ComponentCategory, ComponentMetricId
 from frequenz.sdk.timeseries import Sample
+from frequenz.sdk.timeseries._quantities import Quantity
 from frequenz.sdk.timeseries._resampling import Resampler, ResamplerConfig, Sink, Source
 
 HOST = "microgrid.sandbox.api.frequenz.io"
@@ -33,11 +34,11 @@ async def _calculate_average(source: Source, sink: Sink) -> None:
         count += 1
         if sample.value is None:
             continue
-        avg = avg * (count - 1) / count + sample.value / count
-        await sink(Sample(datetime.now(timezone.utc), avg))
+        avg = avg * (count - 1) / count + sample.value.base_value / count
+        await sink(Sample(datetime.now(timezone.utc), Quantity(avg)))
 
 
-async def _print_sample(sample: Sample) -> None:
+async def _print_sample(sample: Sample[Quantity]) -> None:
     print(f"\nResampled average at {sample.timestamp}: {sample.value}\n")
 
 
@@ -105,7 +106,7 @@ async def run() -> None:  # pylint: disable=too-many-locals
     )
 
     # Create a channel to calculate an average for all the data
-    average_chan = Broadcast[Sample]("average")
+    average_chan = Broadcast[Sample[Quantity]]("average")
 
     second_stage_resampler = Resampler(
         ResamplerConfig(resampling_period=timedelta(seconds=3.0))
@@ -117,7 +118,7 @@ async def run() -> None:  # pylint: disable=too-many-locals
     average_sender = average_chan.new_sender()
 
     # Needed until channels Senders raises exceptions on errors
-    async def sink_adapter(sample: Sample) -> None:
+    async def sink_adapter(sample: Sample[Quantity]) -> None:
         await average_sender.send(sample)
 
     print("Starting...")

pyproject.toml

@@ -113,6 +113,9 @@ disable = [
   # disabled because it conflicts with isort
   "wrong-import-order",
   "ungrouped-imports",
+  # pylint's unsubscriptable check is buggy and is not needed because
+  # it is a type-check, for which we already have mypy.
+  "unsubscriptable-object",
 ]
 
 [tool.pylint.design]

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

@@ -21,6 +21,7 @@
     MeterData,
 )
 from ...timeseries import Sample
+from ...timeseries._quantities import Quantity
 from .._channel_registry import ChannelRegistry
 
 
@@ -304,7 +305,7 @@ def _get_metric_senders(
         self,
         category: ComponentCategory,
         requests: Dict[ComponentMetricId, List[ComponentMetricRequest]],
-    ) -> List[Tuple[Callable[[Any], float], List[Sender[Sample]]]]:
+    ) -> List[Tuple[Callable[[Any], float], List[Sender[Sample[Quantity]]]]]:
         """Get channel senders from the channel registry for each requested metric.
 
         Args:
@@ -355,7 +356,9 @@ def process_msg(data: Any) -> None:
             tasks = []
             for extractor, senders in stream_senders:
                 for sender in senders:
-                    tasks.append(sender.send(Sample(data.timestamp, extractor(data))))
+                    tasks.append(
+                        sender.send(Sample(data.timestamp, Quantity(extractor(data))))
+                    )
             asyncio.gather(*tasks)
             nonlocal pending_messages
             pending_messages -= 1

src/frequenz/sdk/actor/_resampling.py

@@ -13,6 +13,7 @@
 
 from .._internal._asyncio import cancel_and_await
 from ..timeseries import Sample
+from ..timeseries._quantities import Quantity
 from ..timeseries._resampling import Resampler, ResamplerConfig, ResamplingError
 from ._channel_registry import ChannelRegistry
 from ._data_sourcing import ComponentMetricRequest
@@ -80,7 +81,7 @@ async def _subscribe(self, request: ComponentMetricRequest) -> None:
         # exceptions to report errors.
         sender = self._channel_registry.new_sender(request.get_channel_name())
 
-        async def sink_adapter(sample: Sample) -> None:
+        async def sink_adapter(sample: Sample[Quantity]) -> None:
             await sender.send(sample)
 
         self._resampler.add_timeseries(request_channel_name, receiver, sink_adapter)