WIP Quantity ABC

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
 
@@ -318,7 +317,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:
@@ -335,7 +334,7 @@ def _get_metric_senders(
                 self._get_data_extraction_method(category, metric),
                 [
                     self._registry.get_or_create(
-                        Sample[Quantity], request.get_channel_name()
+                        Sample[float], request.get_channel_name()
                     ).new_sender()
                     for request in req_list
                 ],
@@ -353,6 +352,9 @@ 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 streaming data.
         """
         try:
             stream_senders = []
@@ -373,9 +375,7 @@ def process_msg(data: Any) -> None:
                 for extractor, senders in stream_senders:
                     for sender in senders:
                         tasks.append(
-                            sender.send(
-                                Sample(data.timestamp, Quantity(extractor(data)))
-                            )
+                            sender.send(Sample(data.timestamp, extractor(data)))
                         )
                 asyncio.gather(*tasks)
                 nonlocal pending_messages

src/frequenz/sdk/actor/_resampling.py

@@ -7,12 +7,12 @@
 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._base_types import Sample
 from ..timeseries._resampling import Resampler, ResamplerConfig, ResamplingError
 from ._actor import Actor
 from ._channel_registry import ChannelRegistry
@@ -21,6 +21,17 @@
 _logger = logging.getLogger(__name__)
 
 
+# We need to use the dataclass decorator again because we are making a required
+# attribute optional, so we need the dataclass to re-generate the constructor with the
+# new signature.
+@dataclasses.dataclass(frozen=True)
+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 +41,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.
@@ -49,13 +60,15 @@ def __init__(  # pylint: disable=too-many-arguments
         """
         super().__init__(name=name)
         self._channel_registry: ChannelRegistry = channel_registry
+
         self._data_sourcing_request_sender: Sender[ComponentMetricRequest] = (
             data_sourcing_request_sender
         )
         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:
@@ -78,19 +91,16 @@ async def _subscribe(self, request: ComponentMetricRequest) -> None:
         data_source_channel_name = data_source_request.get_channel_name()
         await self._data_sourcing_request_sender.send(data_source_request)
         receiver = self._channel_registry.get_or_create(
-            Sample[Quantity], data_source_channel_name
+            Sample[float], data_source_channel_name
         ).new_receiver()
 
         # This is a temporary hack until the Sender implementation uses
         # exceptions to report errors.
         sender = self._channel_registry.get_or_create(
-            Sample[Quantity], request_channel_name
+            Sample[float], request_channel_name
         ).new_sender()
 
-        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."""
@@ -113,8 +123,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

@@ -121,7 +121,8 @@
 to limit the charge power of individual EV Chargers.
 """  # noqa: D205, D400
 
-from ..actor import ResamplerConfig
+import typing
+
 from . import _data_pipeline, client, component, connection_manager, metadata
 from ._data_pipeline import (
     battery_pool,
@@ -134,8 +135,13 @@
     voltage,
 )
 
+if typing.TYPE_CHECKING:
+    from ..actor._resampling import ResamplingActorConfig
+
 
-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

@@ -35,7 +35,7 @@
 #
 # pylint: disable=import-outside-toplevel
 if typing.TYPE_CHECKING:
-    from ..actor import ComponentMetricRequest, ResamplerConfig, _power_managing
+    from ..actor import ComponentMetricRequest, ResamplingActorConfig, _power_managing
     from ..actor.power_distributing import (  # noqa: F401 (imports used by string type hints)
         ComponentPoolStatus,
         PowerDistributingActor,
@@ -81,7 +81,7 @@ class _DataPipeline:  # pylint: disable=too-many-instance-attributes
 
     def __init__(
         self,
-        resampler_config: ResamplerConfig,
+        resampler_config: ResamplingActorConfig,
     ) -> None:
         """Create a `DataPipeline` instance.
 
@@ -90,7 +90,7 @@ def __init__(
         """
         from ..actor import ChannelRegistry
 
-        self._resampler_config: ResamplerConfig = resampler_config
+        self._resampler_config: ResamplingActorConfig = resampler_config
 
         self._channel_registry: ChannelRegistry = ChannelRegistry(
             name="Data Pipeline Registry"
@@ -408,7 +408,7 @@ async def _stop(self) -> None:
 _DATA_PIPELINE: _DataPipeline | None = None
 
 
-async def initialize(resampler_config: ResamplerConfig) -> None:
+async def initialize(resampler_config: ResamplingActorConfig) -> None:
     """Initialize a `DataPipeline` instance.
 
     Args:

src/frequenz/sdk/timeseries/_base_types.py

@@ -9,16 +9,19 @@
 from collections.abc import Callable, Iterator
 from dataclasses import dataclass
 from datetime import datetime, timezone
-from typing import Generic, Self, TypeVar, overload
+from typing import Generic, Self, SupportsFloat, TypeVar, overload
 
-from ._quantities import Power, QuantityT
+from ._quantities import Power
+
+SupportsFloatT = TypeVar("SupportsFloatT", bound=SupportsFloat)
+"""Type variable for types that support conversion to float."""
 
 UNIX_EPOCH = datetime.fromtimestamp(0.0, tz=timezone.utc)
 """The UNIX epoch (in UTC)."""
 
 
 @dataclass(frozen=True, order=True)
-class Sample(Generic[QuantityT]):
+class Sample(Generic[SupportsFloatT]):
     """A measurement taken at a particular point in time.
 
     The `value` could be `None` if a component is malfunctioning or data is
@@ -29,12 +32,12 @@ class Sample(Generic[QuantityT]):
     timestamp: datetime
     """The time when this sample was generated."""
 
-    value: QuantityT | None = None
+    value: SupportsFloatT | None = None
     """The value of this sample."""
 
 
 @dataclass(frozen=True)
-class Sample3Phase(Generic[QuantityT]):
+class Sample3Phase(Generic[SupportsFloatT]):
     """A 3-phase measurement made at a particular point in time.
 
     Each of the `value` fields could be `None` if a component is malfunctioning
@@ -45,16 +48,16 @@ class Sample3Phase(Generic[QuantityT]):
 
     timestamp: datetime
     """The time when this sample was generated."""
-    value_p1: QuantityT | None
+    value_p1: SupportsFloatT | None
     """The value of the 1st phase in this sample."""
 
-    value_p2: QuantityT | None
+    value_p2: SupportsFloatT | None
     """The value of the 2nd phase in this sample."""
 
-    value_p3: QuantityT | None
+    value_p3: SupportsFloatT | None
     """The value of the 3rd phase in this sample."""
 
-    def __iter__(self) -> Iterator[QuantityT | None]:
+    def __iter__(self) -> Iterator[SupportsFloatT | None]:
         """Return an iterator that yields values from each of the phases.
 
         Yields:
@@ -65,12 +68,12 @@ def __iter__(self) -> Iterator[QuantityT | None]:
         yield self.value_p3
 
     @overload
-    def max(self, default: QuantityT) -> QuantityT: ...
+    def max(self, default: SupportsFloatT) -> SupportsFloatT: ...
 
     @overload
-    def max(self, default: None = None) -> QuantityT | None: ...
+    def max(self, default: None = None) -> SupportsFloatT | None: ...
 
-    def max(self, default: QuantityT | None = None) -> QuantityT | None:
+    def max(self, default: SupportsFloatT | None = None) -> SupportsFloatT | None:
         """Return the max value among all phases, or default if they are all `None`.
 
         Args:
@@ -81,19 +84,19 @@ def max(self, default: QuantityT | None = None) -> QuantityT | None:
         """
         if not any(self):
             return default
-        value: QuantityT = functools.reduce(
-            lambda x, y: x if x > y else y,
+        value: SupportsFloatT = functools.reduce(
+            lambda x, y: x if float(x) > float(y) else y,
             filter(None, self),
         )
         return value
 
     @overload
-    def min(self, default: QuantityT) -> QuantityT: ...
+    def min(self, default: SupportsFloatT) -> SupportsFloatT: ...
 
     @overload
-    def min(self, default: None = None) -> QuantityT | None: ...
+    def min(self, default: None = None) -> SupportsFloatT | None: ...
 
-    def min(self, default: QuantityT | None = None) -> QuantityT | None:
+    def min(self, default: SupportsFloatT | None = None) -> SupportsFloatT | None:
         """Return the min value among all phases, or default if they are all `None`.
 
         Args:
@@ -104,16 +107,16 @@ def min(self, default: QuantityT | None = None) -> QuantityT | None:
         """
         if not any(self):
             return default
-        value: QuantityT = functools.reduce(
-            lambda x, y: x if x < y else y,
+        value: SupportsFloatT = functools.reduce(
+            lambda x, y: x if float(x) < float(y) else y,
             filter(None, self),
         )
         return value
 
     def map(
         self,
-        function: Callable[[QuantityT], QuantityT],
-        default: QuantityT | None = None,
+        function: Callable[[SupportsFloatT], SupportsFloatT],
+        default: SupportsFloatT | None = None,
     ) -> Self:
         """Apply the given function on each of the phase values and return the result.
 

src/frequenz/sdk/timeseries/_grid_frequency.py

@@ -7,6 +7,7 @@
 
 import asyncio
 import logging
+import math
 from typing import TYPE_CHECKING
 
 from frequenz.channels import Receiver, Sender
@@ -15,7 +16,7 @@
 from ..microgrid import connection_manager
 from ..microgrid.component import Component, ComponentCategory, ComponentMetricId
 from ..timeseries._base_types import Sample
-from ..timeseries._quantities import Frequency, Quantity
+from ..timeseries._quantities import Frequency
 
 if TYPE_CHECKING:
     # Imported here to avoid a circular import.
@@ -96,7 +97,7 @@ def new_receiver(self) -> Receiver[Sample[Frequency]]:
             A receiver that will receive grid frequency samples.
         """
         receiver = self._channel_registry.get_or_create(
-            Sample[Quantity], self._component_metric_request.get_channel_name()
+            Sample[float], self._component_metric_request.get_channel_name()
         ).new_receiver()
 
         if not self._task:
@@ -109,10 +110,8 @@ def new_receiver(self) -> Receiver[Sample[Frequency]]:
         return receiver.map(
             lambda sample: (
                 Sample[Frequency](sample.timestamp, None)
-                if sample.value is None or sample.value.isnan()
-                else Sample(
-                    sample.timestamp, Frequency.from_hertz(sample.value.base_value)
-                )
+                if sample.value is None or math.isnan(sample.value)
+                else Sample(sample.timestamp, Frequency.from_hertz(float(sample.value)))
             )
         )