Disable default constructor in specialized Quantity types (#465)

Author: shsms

RELEASE_NOTES.md

@@ -6,12 +6,32 @@
 
 ## Upgrading
 
+- `Sample` objects no longer hold `float`s, but rather `Quantity` or one of its subclasses, like `Power`, `Current`, `Energy`, etc. based on the type of values being streamed.
+
+  ```python
+  sample: Sample[Power] = await battery_pool.power.new_receiver().receive()
+  power: float = sample.value.as_watts()
+  ```
+
+- `BatteryPool.soc` now streams values of type `Sample[Quantity]`, and `BatteryPool.capacity` now streams values of type `Sample[Energy]`.
+
+  ```python
+  battery_pool = microgrid.battery_pool()
+  soc_sample: Sample[Quantity] = await battery_pool.soc.new_receiver().receive()
+  soc: float = soc_sample.value.base_value
+
+  capacity_sample: Sample[Energy] = await battery_pool.capacity.new_receiver().receive()
+  capacity: float = soc_sample.value.as_watt_hours()
+  ```
+
 <!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
 
 ## New Features
 
 - The logical meter has a new method that returns producer power, that is the sum of all energy producers.
 
+- `Quantity` types (`Power`, `Current`, `Energy`, `Voltage`) for providing type- and unit-safety when dealing with physical quantities.
+
 <!-- Here goes the main new features and examples or instructions on how to use them -->
 
 ## Bug Fixes

src/frequenz/sdk/timeseries/_formula_engine/_formula_engine.py

@@ -12,6 +12,7 @@
 from datetime import datetime
 from math import isinf, isnan
 from typing import (
+    Callable,
     Dict,
     Generic,
     List,
@@ -63,23 +64,23 @@ def __init__(
         name: str,
         steps: List[FormulaStep],
         metric_fetchers: Dict[str, MetricFetcher[QuantityT]],
-        output_type: Type[QuantityT],
+        create_method: Callable[[float], QuantityT],
     ) -> None:
         """Create a `FormulaEngine` instance.
 
         Args:
             name: A name for the formula.
             steps: Steps for the engine to execute, in post-fix order.
             metric_fetchers: Fetchers for each metric stream the formula depends on.
-            output_type: A type object to generate the output `Sample` with.  If the
-                formula is for generating power values, this would be `Power`, for
-                example.
+            create_method: A method to generate the output `Sample` value with.  If the
+                formula is for generating power values, this would be
+                `Power.from_watts`, for example.
         """
         self._name = name
         self._steps = steps
         self._metric_fetchers: Dict[str, MetricFetcher[QuantityT]] = metric_fetchers
         self._first_run = True
-        self._output_type: type[QuantityT] = output_type
+        self._create_method: Callable[[float], QuantityT] = create_method
 
     async def _synchronize_metric_timestamps(
         self, metrics: Set[asyncio.Task[Optional[Sample[QuantityT]]]]
@@ -172,7 +173,7 @@ async def apply(self) -> Sample[QuantityT]:
         if isnan(res) or isinf(res):
             return Sample(metric_ts, None)
 
-        return Sample(metric_ts, self._output_type(res))
+        return Sample(metric_ts, self._create_method(res))
 
 
 _CompositionType = Union[
@@ -218,7 +219,7 @@ class _ComposableFormulaEngine(
 ):
     """A base class for formula engines."""
 
-    _output_type: Type[QuantityT]
+    _create_method: Callable[[float], QuantityT]
     _higher_order_builder: Type[_GenericHigherOrderBuilder]
     _task: asyncio.Task[None] | None = None
 
@@ -242,7 +243,7 @@ def __add__(
             A formula builder that can take further expressions, or can be built
                 into a formula engine.
         """
-        return self._higher_order_builder(self, self._output_type) + other  # type: ignore
+        return self._higher_order_builder(self, self._create_method) + other  # type: ignore
 
     def __sub__(
         self, other: _GenericEngine | _GenericHigherOrderBuilder
@@ -257,7 +258,7 @@ def __sub__(
             A formula builder that can take further expressions, or can be built
                 into a formula engine.
         """
-        return self._higher_order_builder(self, self._output_type) - other  # type: ignore
+        return self._higher_order_builder(self, self._create_method) - other  # type: ignore
 
     def __mul__(
         self, other: _GenericEngine | _GenericHigherOrderBuilder
@@ -272,7 +273,7 @@ def __mul__(
             A formula builder that can take further expressions, or can be built
                 into a formula engine.
         """
-        return self._higher_order_builder(self, self._output_type) * other  # type: ignore
+        return self._higher_order_builder(self, self._create_method) * other  # type: ignore
 
     def __truediv__(
         self, other: _GenericEngine | _GenericHigherOrderBuilder
@@ -287,7 +288,7 @@ def __truediv__(
             A formula builder that can take further expressions, or can be built
                 into a formula engine.
         """
-        return self._higher_order_builder(self, self._output_type) / other  # type: ignore
+        return self._higher_order_builder(self, self._create_method) / other  # type: ignore
 
 
 class FormulaEngine(
@@ -307,28 +308,28 @@ class FormulaEngine(
     def __init__(
         self,
         builder: FormulaBuilder[QuantityT],
-        output_type: Type[QuantityT],
+        create_method: Callable[[float], QuantityT],
     ) -> None:
         """Create a `FormulaEngine` instance.
 
         Args:
             builder: A `FormulaBuilder` instance to get the formula steps and metric
                 fetchers from.
-            output_type: A type object to generate the output `Sample` with.  If the
-                formula is for generating power values, this would be `Power`, for
-                example.
+            create_method: A method to generate the output `Sample` value with.  If the
+                formula is for generating power values, this would be
+                `Power.from_watts`, for example.
         """
         self._higher_order_builder = HigherOrderFormulaBuilder
         self._name: str = builder.name
         self._builder: FormulaBuilder[QuantityT] = builder
-        self._output_type: Type[QuantityT] = output_type  # type: ignore
+        self._create_method = create_method
         self._channel: Broadcast[Sample[QuantityT]] = Broadcast(self._name)
 
     async def _run(self) -> None:
         await self._builder.subscribe()
         steps, metric_fetchers = self._builder.finalize()
         evaluator = FormulaEvaluator[QuantityT](
-            self._name, steps, metric_fetchers, self._output_type
+            self._name, steps, metric_fetchers, self._create_method
         )
         sender = self._channel.new_sender()
         while True:
@@ -378,7 +379,7 @@ class FormulaEngine3Phase(
     def __init__(
         self,
         name: str,
-        output_type: Type[QuantityT],
+        create_method: Callable[[float], QuantityT],
         phase_streams: Tuple[
             FormulaEngine[QuantityT],
             FormulaEngine[QuantityT],
@@ -389,14 +390,14 @@ def __init__(
 
         Args:
             name: A name for the formula.
-            output_type: A type object to generate the output `Sample` with.  If the
-                formula is for generating power values, this would be `Power`, for
-                example.
+            create_method: A method to generate the output `Sample` value with.  If the
+                formula is for generating power values, this would be
+                `Power.from_watts`, for example.
             phase_streams: output streams of formula engines running per-phase formulas.
         """
         self._higher_order_builder = HigherOrderFormulaBuilder3Phase
         self._name: str = name
-        self._output_type = output_type
+        self._create_method = create_method
         self._channel: Broadcast[Sample3Phase[QuantityT]] = Broadcast(self._name)
         self._task: asyncio.Task[None] | None = None
         self._streams: tuple[
@@ -474,17 +475,17 @@ class FormulaBuilder(Generic[QuantityT]):
         add the values and return the result.
     """
 
-    def __init__(self, name: str, output_type: Type[QuantityT]) -> None:
+    def __init__(self, name: str, create_method: Callable[[float], QuantityT]) -> None:
         """Create a `FormulaBuilder` instance.
 
         Args:
             name: A name for the formula being built.
-            output_type: A type object to generate the output `Sample` with.  If the
-                formula is for generating power values, this would be `Power`, for
-                example.
+            create_method: A method to generate the output `Sample` value with.  If the
+                formula is for generating power values, this would be
+                `Power.from_watts`, for example.
         """
         self._name = name
-        self._output_type: Type[QuantityT] = output_type
+        self._create_method: Callable[[float], QuantityT] = create_method
         self._build_stack: List[FormulaStep] = []
         self._steps: List[FormulaStep] = []
         self._metric_fetchers: Dict[str, MetricFetcher[QuantityT]] = {}
@@ -648,7 +649,7 @@ def build(self) -> FormulaEngine[QuantityT]:
             A `FormulaEngine` instance.
         """
         self.finalize()
-        return FormulaEngine(self, output_type=self._output_type)
+        return FormulaEngine(self, create_method=self._create_method)
 
 
 class _BaseHOFormulaBuilder(ABC, Generic[QuantityT]):
@@ -657,16 +658,16 @@ class _BaseHOFormulaBuilder(ABC, Generic[QuantityT]):
     def __init__(
         self,
         engine: FormulaEngine[QuantityT] | FormulaEngine3Phase[QuantityT],
-        output_type: Type[QuantityT],
+        create_method: Callable[[float], QuantityT],
     ) -> None:
         """Create a `GenericHigherOrderFormulaBuilder` instance.
 
         Args:
             engine: A first input stream to create a builder with, so that python
                 operators `+, -, *, /` can be used directly on newly created instances.
-            output_type: A type object to generate the output `Sample` with.  If the
-                formula is for generating power values, this would be `Power`, for
-                example.
+            create_method: A method to generate the output `Sample` value with.  If the
+                formula is for generating power values, this would be
+                `Power.from_watts`, for example.
         """
         self._steps: deque[
             tuple[
@@ -675,7 +676,7 @@ def __init__(
             ]
         ] = deque()
         self._steps.append((TokenType.COMPONENT_METRIC, engine))
-        self._output_type: Type[QuantityT] = output_type
+        self._create_method: Callable[[float], QuantityT] = create_method
 
     @overload
     def _push(
@@ -854,7 +855,7 @@ def build(
         Returns:
             A `FormulaEngine` instance.
         """
-        builder = FormulaBuilder(name, self._output_type)
+        builder = FormulaBuilder(name, self._create_method)
         for typ, value in self._steps:
             if typ == TokenType.COMPONENT_METRIC:
                 assert isinstance(value, FormulaEngine)
@@ -888,9 +889,9 @@ def build(
             A `FormulaEngine3Phase` instance.
         """
         builders = [
-            FormulaBuilder(name, self._output_type),
-            FormulaBuilder(name, self._output_type),
-            FormulaBuilder(name, self._output_type),
+            FormulaBuilder(name, self._create_method),
+            FormulaBuilder(name, self._create_method),
+            FormulaBuilder(name, self._create_method),
         ]
         for typ, value in self._steps:
             if typ == TokenType.COMPONENT_METRIC:
@@ -909,7 +910,7 @@ def build(
                     builders[phase].push_oper(value)
         return FormulaEngine3Phase(
             name,
-            self._output_type,
+            self._create_method,
             (
                 builders[0].build(),
                 builders[1].build(),

src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_battery_power_formula.py

@@ -43,7 +43,7 @@ def generate(
                 in the component graph.
         """
         builder = self._get_builder(
-            "battery-power", ComponentMetricId.ACTIVE_POWER, Power
+            "battery-power", ComponentMetricId.ACTIVE_POWER, Power.from_watts
         )
         component_ids = self._config.component_ids
         if not component_ids:

src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_chp_power_formula.py

@@ -38,7 +38,9 @@ def generate(self) -> FormulaEngine[Power]:
             FormulaGenerationError: If there's no dedicated meter attached to every CHP.
 
         """
-        builder = self._get_builder("chp-power", ComponentMetricId.ACTIVE_POWER, Power)
+        builder = self._get_builder(
+            "chp-power", ComponentMetricId.ACTIVE_POWER, Power.from_watts
+        )
 
         chp_meter_ids = self._get_chp_meters()
         if not chp_meter_ids:

src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_consumer_power_formula.py

@@ -35,7 +35,7 @@ def generate(self) -> FormulaEngine[Power]:
                 meter.
         """
         builder = self._get_builder(
-            "consumer-power", ComponentMetricId.ACTIVE_POWER, Power
+            "consumer-power", ComponentMetricId.ACTIVE_POWER, Power.from_watts
         )
         component_graph = connection_manager.get().component_graph
         grid_component = next(

src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_ev_charger_current_formula.py

@@ -37,21 +37,21 @@ def generate(self) -> FormulaEngine3Phase[Current]:
             # frequency as the other streams.  So we subscribe with a non-existing
             # component id, just to get a `None` message at the resampling interval.
             builder = self._get_builder(
-                "ev-current", ComponentMetricId.ACTIVE_POWER, Current
+                "ev-current", ComponentMetricId.ACTIVE_POWER, Current.from_amperes
             )
             builder.push_component_metric(
                 NON_EXISTING_COMPONENT_ID, nones_are_zeros=True
             )
             engine = builder.build()
             return FormulaEngine3Phase(
                 "ev-current",
-                Current,
+                Current.from_amperes,
                 (engine, engine, engine),
             )
 
         return FormulaEngine3Phase(
             "ev-current",
-            Current,
+            Current.from_amperes,
             (
                 (
                     self._gen_phase_formula(
@@ -76,7 +76,7 @@ def _gen_phase_formula(
         component_ids: abc.Set[int],
         metric_id: ComponentMetricId,
     ) -> FormulaEngine[Current]:
-        builder = self._get_builder("ev-current", metric_id, Current)
+        builder = self._get_builder("ev-current", metric_id, Current.from_amperes)
 
         # generate a formula that just adds values from all EV Chargers.
         for idx, component_id in enumerate(component_ids):

src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_ev_charger_power_formula.py

@@ -22,7 +22,9 @@ def generate(self) -> FormulaEngine[Power]:
         Returns:
             A formula engine that calculates total EV Charger power values.
         """
-        builder = self._get_builder("ev-power", ComponentMetricId.ACTIVE_POWER, Power)
+        builder = self._get_builder(
+            "ev-power", ComponentMetricId.ACTIVE_POWER, Power.from_watts
+        )
 
         component_ids = self._config.component_ids
         if not component_ids:

src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_formula_generator.py

@@ -10,7 +10,7 @@
 from collections import abc
 from dataclasses import dataclass
 from enum import Enum
-from typing import Generic, Type
+from typing import Callable, Generic
 
 from frequenz.channels import Sender
 
@@ -88,15 +88,15 @@ def _get_builder(
         self,
         name: str,
         component_metric_id: ComponentMetricId,
-        output_type: Type[QuantityT],
+        create_method: Callable[[float], QuantityT],
     ) -> ResampledFormulaBuilder[QuantityT]:
         builder = ResampledFormulaBuilder(
             self._namespace,
             name,
             self._channel_registry,
             self._resampler_subscription_sender,
             component_metric_id,
-            output_type,
+            create_method,
         )
         return builder
 

src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_grid_current_formula.py

@@ -43,7 +43,7 @@ def generate(self) -> FormulaEngine3Phase[Current]:
 
         return FormulaEngine3Phase(
             "grid-current",
-            Current,
+            Current.from_amperes,
             (
                 self._gen_phase_formula(
                     grid_successors, ComponentMetricId.CURRENT_PHASE_1
@@ -62,7 +62,7 @@ def _gen_phase_formula(
         grid_successors: Set[Component],
         metric_id: ComponentMetricId,
     ) -> FormulaEngine[Current]:
-        builder = self._get_builder("grid-current", metric_id, Current)
+        builder = self._get_builder("grid-current", metric_id, Current.from_amperes)
 
         # generate a formula that just adds values from all components that are
         # directly connected to the grid.