Make Samples hold Quantity instances instead of float

This makes the transition to a generic `Sample` type easier, which is
done in a subsequent commit.

This commit also introduces the `output_type` parameter to formula
builders and formula engines.  The new parameter is used to determine
which type of quantity instance a formula engine would produce.

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

Author: shsms

benchmarks/timeseries/periodic_feature_extractor.py

@@ -24,6 +24,7 @@
 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:
@@ -34,7 +35,7 @@ async def init_feature_extractor(period: int) -> PeriodicFeatureExtractor:
         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,
@@ -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

@@ -126,7 +126,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,8 +34,8 @@ 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:

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
 
 
@@ -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/timeseries/_base_types.py

@@ -8,6 +8,8 @@
 from datetime import datetime, timezone
 from typing import Callable, Iterator, Self, overload
 
+from ._quantities import Quantity
+
 UNIX_EPOCH = datetime.fromtimestamp(0.0, tz=timezone.utc)
 """The UNIX epoch (in UTC)."""
 
@@ -24,7 +26,7 @@ class Sample:
     timestamp: datetime
     """The time when this sample was generated."""
 
-    value: float | None = None
+    value: Quantity | None = None
     """The value of this sample."""
 
 
@@ -40,16 +42,16 @@ class Sample3Phase:
 
     timestamp: datetime
     """The time when this sample was generated."""
-    value_p1: float | None
+    value_p1: Quantity | None
     """The value of the 1st phase in this sample."""
 
-    value_p2: float | None
+    value_p2: Quantity | None
     """The value of the 2nd phase in this sample."""
 
-    value_p3: float | None
+    value_p3: Quantity | None
     """The value of the 3rd phase in this sample."""
 
-    def __iter__(self) -> Iterator[float | None]:
+    def __iter__(self) -> Iterator[Quantity | None]:
         """Return an iterator that yields values from each of the phases.
 
         Yields:
@@ -60,14 +62,14 @@ def __iter__(self) -> Iterator[float | None]:
         yield self.value_p3
 
     @overload
-    def max(self, default: float) -> float:
+    def max(self, default: Quantity) -> Quantity:
         ...
 
     @overload
-    def max(self, default: None = None) -> float | None:
+    def max(self, default: None = None) -> Quantity | None:
         ...
 
-    def max(self, default: float | None = None) -> float | None:
+    def max(self, default: Quantity | None = None) -> Quantity | None:
         """Return the max value among all phases, or default if they are all `None`.
 
         Args:
@@ -78,21 +80,21 @@ def max(self, default: float | None = None) -> float | None:
         """
         if not any(self):
             return default
-        value: float = functools.reduce(
+        value: Quantity = functools.reduce(
             lambda x, y: x if x > y else y,
             filter(None, self),
         )
         return value
 
     @overload
-    def min(self, default: float) -> float:
+    def min(self, default: Quantity) -> Quantity:
         ...
 
     @overload
-    def min(self, default: None = None) -> float | None:
+    def min(self, default: None = None) -> Quantity | None:
         ...
 
-    def min(self, default: float | None = None) -> float | None:
+    def min(self, default: Quantity | None = None) -> Quantity | None:
         """Return the min value among all phases, or default if they are all `None`.
 
         Args:
@@ -103,14 +105,16 @@ def min(self, default: float | None = None) -> float | None:
         """
         if not any(self):
             return default
-        value: float = functools.reduce(
+        value: Quantity = functools.reduce(
             lambda x, y: x if x < y else y,
             filter(None, self),
         )
         return value
 
     def map(
-        self, function: Callable[[float], float], default: float | None = None
+        self,
+        function: Callable[[Quantity], Quantity],
+        default: Quantity | None = None,
     ) -> Self:
         """Apply the given function on each of the phase values and return the result.