WIP: quantity try 2

Author: llucax

src/frequenz/sdk/actor/__init__.py

@@ -592,13 +592,12 @@ async def main() -> None:  # (6)!
 [_run]: #the-_run-method
 """
 
-from ..timeseries._resampling import ResamplerConfig
 from ._actor import Actor
 from ._background_service import BackgroundService
 from ._channel_registry import ChannelRegistry
 from ._config_managing import ConfigManagingActor
 from ._data_sourcing import ComponentMetricRequest, DataSourcingActor
-from ._resampling import ComponentMetricsResamplingActor
+from ._resampling import ComponentMetricsResamplingActor, ResamplingActorConfig
 from ._run_utils import run
 
 __all__ = [
@@ -609,6 +608,6 @@ async def main() -> None:  # (6)!
     "ComponentMetricsResamplingActor",
     "ConfigManagingActor",
     "DataSourcingActor",
-    "ResamplerConfig",
+    "ResamplingActorConfig",
     "run",
 ]

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

@@ -20,7 +20,6 @@
     MeterData,
 )
 from ...timeseries import Sample
-from ...timeseries._quantities import Quantity
 from .._channel_registry import ChannelRegistry
 from ._component_metric_request import ComponentMetricRequest
 
@@ -306,7 +305,7 @@ def _get_metric_senders(
         self,
         category: ComponentCategory,
         requests: dict[ComponentMetricId, list[ComponentMetricRequest]],
-    ) -> list[tuple[Callable[[Any], float], list[Sender[Sample[Quantity]]]]]:
+    ) -> list[tuple[Callable[[Any], float], list[Sender[Sample[float]]]]]:
         """Get channel senders from the channel registry for each requested metric.
 
         Args:
@@ -357,9 +356,7 @@ 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))))
-                    )
+                    tasks.append(sender.send(Sample(data.timestamp, extractor(data))))
             asyncio.gather(*tasks)
             nonlocal pending_messages
             pending_messages -= 1

src/frequenz/sdk/actor/_resampling.py

@@ -7,12 +7,11 @@
 import asyncio
 import dataclasses
 import logging
+from typing import Callable
 
 from frequenz.channels import Receiver, Sender
 
 from .._internal._asyncio import cancel_and_await
-from ..timeseries import Sample
-from ..timeseries._quantities import Quantity
 from ..timeseries._resampling import Resampler, ResamplerConfig, ResamplingError
 from ._actor import Actor
 from ._channel_registry import ChannelRegistry
@@ -21,6 +20,13 @@
 _logger = logging.getLogger(__name__)
 
 
+class ResamplingActorConfig(ResamplerConfig[float]):
+    """Configuration for the resampling actor."""
+
+    value_constructor: Callable[[float], float] = float
+    """The constructor to use to create new sample values."""
+
+
 class ComponentMetricsResamplingActor(Actor):
     """An actor to resample microgrid component metrics."""
 
@@ -30,7 +36,7 @@ def __init__(  # pylint: disable=too-many-arguments
         channel_registry: ChannelRegistry,
         data_sourcing_request_sender: Sender[ComponentMetricRequest],
         resampling_request_receiver: Receiver[ComponentMetricRequest],
-        config: ResamplerConfig,
+        config: ResamplingActorConfig,
         name: str | None = None,
     ) -> None:
         """Initialize an instance.
@@ -55,7 +61,7 @@ def __init__(  # pylint: disable=too-many-arguments
         self._resampling_request_receiver: Receiver[
             ComponentMetricRequest
         ] = resampling_request_receiver
-        self._resampler: Resampler = Resampler(config)
+        self._resampler: Resampler[float] = Resampler(config)
         self._active_req_channels: set[str] = set()
 
     async def _subscribe(self, request: ComponentMetricRequest) -> None:
@@ -83,10 +89,7 @@ async def _subscribe(self, request: ComponentMetricRequest) -> None:
         # exceptions to report errors.
         sender = self._channel_registry.new_sender(request_channel_name)
 
-        async def sink_adapter(sample: Sample[Quantity]) -> None:
-            await sender.send(sample)
-
-        self._resampler.add_timeseries(request_channel_name, receiver, sink_adapter)
+        self._resampler.add_timeseries(request_channel_name, receiver, sender.send)
 
     async def _process_resampling_requests(self) -> None:
         """Process resampling data requests."""
@@ -109,8 +112,6 @@ async def _run(self) -> None:
         Raises:
             RuntimeError: If there is some unexpected error while resampling or
                 handling requests.
-
-        [//]: # (# noqa: DAR401 error)
         """
         tasks_to_cancel: set[asyncio.Task[None]] = set()
         try:

src/frequenz/sdk/microgrid/__init__.py

@@ -122,7 +122,7 @@
 to limit the charge power of individual EV Chargers.
 """  # noqa: D205, D400
 
-from ..actor import ResamplerConfig
+from ..actor import ResamplingActorConfig
 from . import _data_pipeline, client, component, connection_manager, metadata
 from ._data_pipeline import (
     battery_pool,
@@ -133,7 +133,9 @@
 )
 
 
-async def initialize(host: str, port: int, resampler_config: ResamplerConfig) -> None:
+async def initialize(
+    host: str, port: int, resampler_config: ResamplingActorConfig
+) -> None:
     """Initialize the microgrid connection manager and the data pipeline.
 
     Args:

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -78,7 +78,7 @@ class _DataPipeline:  # pylint: disable=too-many-instance-attributes
 
     def __init__(
         self,
-        resampler_config: ResamplerConfig,
+        resampler_config: ResamplerConfig[float],
     ) -> None:
         """Create a `DataPipeline` instance.
 
@@ -384,7 +384,7 @@ async def _stop(self) -> None:
 _DATA_PIPELINE: _DataPipeline | None = None
 
 
-async def initialize(resampler_config: ResamplerConfig) -> None:
+async def initialize(resampler_config: ResamplerConfig[float]) -> None:
     """Initialize a `DataPipeline` instance.
 
     Args:

src/frequenz/sdk/timeseries/_base_types.py

@@ -10,16 +10,37 @@
 from collections.abc import Callable, Iterator
 from dataclasses import dataclass
 from datetime import datetime, timezone
-from typing import Generic, Self, overload
+from typing import Generic, Protocol, Self, SupportsFloat, overload
 
-from ._quantities import Power, QuantityT
+from ._quantities import Power
 
 UNIX_EPOCH = datetime.fromtimestamp(0.0, tz=timezone.utc)
 """The UNIX epoch (in UTC)."""
 
 
+class Comparable(Protocol):
+    def __lt__(self, other: Self) -> bool:
+        ...
+
+    def __gt__(self, other: Self) -> bool:
+        ...
+
+    def __le__(self, other: Self) -> bool:
+        ...
+
+    def __ge__(self, other: Self) -> bool:
+        ...
+
+
+_T = typing.TypeVar("_T")
+SupportsFloatT = typing.TypeVar("SupportsFloatT", bound=SupportsFloat)
+"""Type variable for types that support conversion to float."""
+
+ComparableT = typing.TypeVar("ComparableT", bound=Comparable)
+
+
 @dataclass(frozen=True, order=True)
-class Sample(Generic[QuantityT]):
+class Sample(Generic[_T]):
     """A measurement taken at a particular point in time.
 
     The `value` could be `None` if a component is malfunctioning or data is
@@ -30,12 +51,12 @@ class Sample(Generic[QuantityT]):
     timestamp: datetime
     """The time when this sample was generated."""
 
-    value: QuantityT | None = None
+    value: _T | None = None
     """The value of this sample."""
 
 
 @dataclass(frozen=True)
-class Sample3Phase(Generic[QuantityT]):
+class Sample3Phase(Generic[ComparableT]):
     """A 3-phase measurement made at a particular point in time.
 
     Each of the `value` fields could be `None` if a component is malfunctioning
@@ -46,16 +67,16 @@ class Sample3Phase(Generic[QuantityT]):
 
     timestamp: datetime
     """The time when this sample was generated."""
-    value_p1: QuantityT | None
+    value_p1: ComparableT | None
     """The value of the 1st phase in this sample."""
 
-    value_p2: QuantityT | None
+    value_p2: ComparableT | None
     """The value of the 2nd phase in this sample."""
 
-    value_p3: QuantityT | None
+    value_p3: ComparableT | None
     """The value of the 3rd phase in this sample."""
 
-    def __iter__(self) -> Iterator[QuantityT | None]:
+    def __iter__(self) -> Iterator[ComparableT | None]:
         """Return an iterator that yields values from each of the phases.
 
         Yields:
@@ -66,14 +87,14 @@ def __iter__(self) -> Iterator[QuantityT | None]:
         yield self.value_p3
 
     @overload
-    def max(self, default: QuantityT) -> QuantityT:
+    def max(self, default: ComparableT) -> ComparableT:
         ...
 
     @overload
-    def max(self, default: None = None) -> QuantityT | None:
+    def max(self, default: None = None) -> ComparableT | None:
         ...
 
-    def max(self, default: QuantityT | None = None) -> QuantityT | None:
+    def max(self, default: ComparableT | None = None) -> ComparableT | None:
         """Return the max value among all phases, or default if they are all `None`.
 
         Args:
@@ -84,21 +105,21 @@ def max(self, default: QuantityT | None = None) -> QuantityT | None:
         """
         if not any(self):
             return default
-        value: QuantityT = functools.reduce(
+        value: ComparableT = functools.reduce(
             lambda x, y: x if x > y else y,
             filter(None, self),
         )
         return value
 
     @overload
-    def min(self, default: QuantityT) -> QuantityT:
+    def min(self, default: ComparableT) -> ComparableT:
         ...
 
     @overload
-    def min(self, default: None = None) -> QuantityT | None:
+    def min(self, default: None = None) -> ComparableT | None:
         ...
 
-    def min(self, default: QuantityT | None = None) -> QuantityT | None:
+    def min(self, default: ComparableT | None = None) -> ComparableT | None:
         """Return the min value among all phases, or default if they are all `None`.
 
         Args:
@@ -109,16 +130,16 @@ def min(self, default: QuantityT | None = None) -> QuantityT | None:
         """
         if not any(self):
             return default
-        value: QuantityT = functools.reduce(
+        value: ComparableT = functools.reduce(
             lambda x, y: x if x < y else y,
             filter(None, self),
         )
         return value
 
     def map(
         self,
-        function: Callable[[QuantityT], QuantityT],
-        default: QuantityT | None = None,
+        function: Callable[[ComparableT], ComparableT],
+        default: ComparableT | None = None,
     ) -> Self:
         """Apply the given function on each of the phase values and return the result.
 
@@ -140,9 +161,6 @@ def map(
         )
 
 
-_T = typing.TypeVar("_T")
-
-
 @dataclass(frozen=True)
 class Bounds(Generic[_T]):
     """Lower and upper bound values."""

src/frequenz/sdk/timeseries/_moving_window.py

@@ -9,22 +9,22 @@
 import math
 from collections.abc import Sequence
 from datetime import datetime, timedelta
-from typing import SupportsIndex, overload
+from typing import Generic, SupportsIndex, overload
 
 import numpy as np
 from frequenz.channels import Broadcast, Receiver, Sender
 from numpy.typing import ArrayLike
 
 from ..actor._background_service import BackgroundService
 from ._base_types import UNIX_EPOCH, Sample
-from ._quantities import Quantity
+from ._quantities import QuantityT
 from ._resampling import Resampler, ResamplerConfig
 from ._ringbuffer import OrderedRingBuffer
 
 _logger = logging.getLogger(__name__)
 
 
-class MovingWindow(BackgroundService):
+class MovingWindow(BackgroundService, Generic[QuantityT]):
     """
     A data window that moves with the latest datapoints of a data stream.
 
@@ -130,9 +130,9 @@ async def run() -> None:
     def __init__(  # pylint: disable=too-many-arguments
         self,
         size: timedelta,
-        resampled_data_recv: Receiver[Sample[Quantity]],
+        resampled_data_recv: Receiver[Sample[QuantityT]],
         input_sampling_period: timedelta,
-        resampler_config: ResamplerConfig | None = None,
+        resampler_config: ResamplerConfig[QuantityT] | None = None,
         align_to: datetime = UNIX_EPOCH,
         *,
         name: str | None = None,
@@ -166,8 +166,8 @@ def __init__(  # pylint: disable=too-many-arguments
 
         self._sampling_period = input_sampling_period
 
-        self._resampler: Resampler | None = None
-        self._resampler_sender: Sender[Sample[Quantity]] | None = None
+        self._resampler: Resampler[QuantityT] | None = None
+        self._resampler_sender: Sender[Sample[QuantityT]] | None = None
 
         if resampler_config:
             assert (
@@ -182,7 +182,7 @@ def __init__(  # pylint: disable=too-many-arguments
             size.total_seconds() / self._sampling_period.total_seconds()
         )
 
-        self._resampled_data_recv = resampled_data_recv
+        self._resampled_data_recv: Receiver[Sample[QuantityT]] = resampled_data_recv
         self._buffer = OrderedRingBuffer(
             np.empty(shape=num_samples, dtype=float),
             sampling_period=self._sampling_period,
@@ -341,14 +341,16 @@ def _configure_resampler(self) -> None:
         """Configure the components needed to run the resampler."""
         assert self._resampler is not None
 
-        async def sink_buffer(sample: Sample[Quantity]) -> None:
+        async def sink_buffer(sample: Sample[QuantityT]) -> None:
             if sample.value is not None:
                 self._buffer.update(sample)
 
-        resampler_channel = Broadcast[Sample[Quantity]]("average")
+        resampler_channel = Broadcast[Sample[QuantityT]]("average")
         self._resampler_sender = resampler_channel.new_sender()
         self._resampler.add_timeseries(
-            "avg", resampler_channel.new_receiver(), sink_buffer
+            "avg",
+            resampler_channel.new_receiver(),
+            sink_buffer,
         )
         self._tasks.add(
             asyncio.create_task(self._resampler.resample(), name="resample")