A Percentage quantity (#470)

Percentage is is a unitless quantity, but with a symbol, and with
multiple representations (as pct, or as a fractional value), so it
somehow fits as a `Quantity`. The biggest benefit would allow for better
user experience for `BatteryPool.soc`, for example:

```python
soc_sample = await battery_pool.soc.new_receiver().receive()
soc_pct = soc_sample.value.base_value   ## old
soc_pct = soc_sample.value.as_percent()  ## new

Author: shsms

src/frequenz/sdk/timeseries/__init__.py

@@ -38,7 +38,7 @@
 from ._base_types import UNIX_EPOCH, Sample, Sample3Phase
 from ._moving_window import MovingWindow
 from ._periodic_feature_extractor import PeriodicFeatureExtractor
-from ._quantities import Current, Energy, Power, Quantity, QuantityT, Voltage
+from ._quantities import Current, Energy, Percentage, Power, Quantity, Voltage
 from ._resampling import ResamplerConfig
 
 __all__ = [
@@ -56,4 +56,5 @@
     "Energy",
     "Power",
     "Voltage",
+    "Percentage",
 ]

src/frequenz/sdk/timeseries/_quantities.py

@@ -9,7 +9,9 @@
 from datetime import timedelta
 from typing import Any, NoReturn, Self, TypeVar, overload
 
-QuantityT = TypeVar("QuantityT", "Quantity", "Power", "Current", "Voltage", "Energy")
+QuantityT = TypeVar(
+    "QuantityT", "Quantity", "Power", "Current", "Voltage", "Energy", "Percentage"
+)
 
 
 class Quantity:
@@ -704,3 +706,55 @@ def __truediv__(self, other: timedelta | Power) -> Power | timedelta:
         raise TypeError(
             f"unsupported operand type(s) for /: '{type(self)}' and '{type(other)}'"
         )
+
+
+class Percentage(
+    Quantity,
+    metaclass=_NoDefaultConstructible,
+    exponent_unit_map={0: "%"},
+):
+    """A percentage quantity."""
+
+    @classmethod
+    def from_percent(cls, percent: float) -> Self:
+        """Initialize a new percentage quantity from a percent value.
+
+        Args:
+            percent: The percent value, normally in the 0.0-100.0 range.
+
+        Returns:
+            A new percentage quantity.
+        """
+        percentage = cls.__new__(cls)
+        percentage._base_value = percent
+        return percentage
+
+    @classmethod
+    def from_fraction(cls, fraction: float) -> Self:
+        """Initialize a new percentage quantity from a fraction.
+
+        Args:
+            fraction: The fraction, normally in the 0.0-1.0 range.
+
+        Returns:
+            A new percentage quantity.
+        """
+        percentage = cls.__new__(cls)
+        percentage._base_value = fraction * 100
+        return percentage
+
+    def as_percent(self) -> float:
+        """Return this quantity as a percentage.
+
+        Returns:
+            This quantity as a percentage.
+        """
+        return self._base_value
+
+    def as_fraction(self) -> float:
+        """Return this quantity as a fraction.
+
+        Returns:
+            This quantity as a fraction.
+        """
+        return self._base_value / 100

src/frequenz/sdk/timeseries/battery_pool/_metric_calculator.py

@@ -12,7 +12,7 @@
 
 from ...microgrid import connection_manager
 from ...microgrid.component import ComponentCategory, ComponentMetricId, InverterType
-from ...timeseries import Energy, Quantity, Sample
+from ...timeseries import Energy, Percentage, Sample
 from ._component_metrics import ComponentMetricsData
 from ._result_types import Bound, PowerMetrics
 
@@ -59,7 +59,7 @@ def battery_inverter_mapping(batteries: Iterable[int]) -> dict[int, int]:
 
 # Formula output types class have no common interface
 # Print all possible types here.
-T = TypeVar("T", Sample[Quantity], Sample[Energy], PowerMetrics)
+T = TypeVar("T", Sample[Percentage], Sample[Energy], PowerMetrics)
 
 
 class MetricCalculator(ABC, Generic[T]):
@@ -234,7 +234,7 @@ def calculate(
         )
 
 
-class SoCCalculator(MetricCalculator[Sample[Quantity]]):
+class SoCCalculator(MetricCalculator[Sample[Percentage]]):
     """Define how to calculate SoC metrics."""
 
     def __init__(self, batteries: Set[int]) -> None:
@@ -283,7 +283,7 @@ def calculate(
         self,
         metrics_data: dict[int, ComponentMetricsData],
         working_batteries: set[int],
-    ) -> Sample[Quantity] | None:
+    ) -> Sample[Percentage] | None:
         """Aggregate the metrics_data and calculate high level metric.
 
         Missing components will be ignored. Formula will be calculated for all
@@ -349,11 +349,11 @@ def calculate(
         if total_capacity_x100 == 0:
             return Sample(
                 timestamp=timestamp,
-                value=Quantity(0.0),
+                value=Percentage.from_percent(0.0),
             )
         return Sample(
             timestamp=timestamp,
-            value=Quantity(used_capacity_x100 / total_capacity_x100),
+            value=Percentage.from_percent(used_capacity_x100 / total_capacity_x100),
         )
 
 

src/frequenz/sdk/timeseries/battery_pool/battery_pool.py

@@ -20,14 +20,14 @@
 from ...actor.power_distributing.result import Result
 from ...microgrid import connection_manager
 from ...microgrid.component import ComponentCategory
-from ...timeseries import Quantity, Sample
+from ...timeseries import Sample
 from .._formula_engine import FormulaEngine, FormulaEnginePool
 from .._formula_engine._formula_generators import (
     BatteryPowerFormula,
     FormulaGeneratorConfig,
     FormulaType,
 )
-from .._quantities import Energy, Power
+from .._quantities import Energy, Percentage, Power
 from ._methods import MetricAggregator, SendOnUpdate
 from ._metric_calculator import CapacityCalculator, PowerBoundsCalculator, SoCCalculator
 from ._result_types import PowerMetrics
@@ -330,7 +330,7 @@ def consumption_power(self) -> FormulaEngine[Power]:
         return engine
 
     @property
-    def soc(self) -> MetricAggregator[Sample[Quantity]]:
+    def soc(self) -> MetricAggregator[Sample[Percentage]]:
         """Fetch the normalized average weighted-by-capacity SoC values for the pool.
 
         The values are normalized to the 0-100% range.

tests/timeseries/_battery_pool/test_battery_pool.py

@@ -24,7 +24,7 @@
 from frequenz.sdk.actor import ResamplerConfig
 from frequenz.sdk.actor.power_distributing import BatteryStatus
 from frequenz.sdk.microgrid.component import ComponentCategory
-from frequenz.sdk.timeseries import Energy, Power, Quantity, Sample
+from frequenz.sdk.timeseries import Energy, Percentage, Power, Sample
 from frequenz.sdk.timeseries.battery_pool import BatteryPool, Bound, PowerMetrics
 from frequenz.sdk.timeseries.battery_pool._metric_calculator import (
     battery_inverter_mapping,
@@ -688,29 +688,29 @@ async def run_soc_test(setup_args: SetupArgs) -> None:
     now = datetime.now(tz=timezone.utc)
     expected = Sample(
         timestamp=now,
-        value=Quantity(10.0),
+        value=Percentage.from_percent(10.0),
     )
     compare_messages(msg, expected, WAIT_FOR_COMPONENT_DATA_SEC + 0.2)
 
     batteries_in_pool = list(battery_pool.battery_ids)
-    scenarios: list[Scenario[Sample[Quantity]]] = [
+    scenarios: list[Scenario[Sample[Percentage]]] = [
         Scenario(
             batteries_in_pool[0],
             {"capacity": 150, "soc": 10},
-            Sample(now, Quantity(2.5)),
+            Sample(now, Percentage.from_percent(2.5)),
         ),
         Scenario(
             batteries_in_pool[0],
             {
                 "soc_lower_bound": 0.0,
             },
-            Sample(now, Quantity(12.727272727272727)),
+            Sample(now, Percentage.from_percent(12.727272727272727)),
         ),
         # If NaN, then not include that battery in the metric.
         Scenario(
             batteries_in_pool[0],
             {"soc_upper_bound": float("NaN")},
-            Sample(now, Quantity(10.0)),
+            Sample(now, Percentage.from_percent(10.0)),
         ),
         # All batteries are sending NaN, can't calculate SoC so we should send None
         Scenario(
@@ -721,7 +721,7 @@ async def run_soc_test(setup_args: SetupArgs) -> None:
         Scenario(
             batteries_in_pool[1],
             {"soc": 30},
-            Sample(now, Quantity(10.0)),
+            Sample(now, Percentage.from_percent(10.0)),
         ),
         # Final metric didn't change, so nothing should be received.
         Scenario(
@@ -734,17 +734,17 @@ async def run_soc_test(setup_args: SetupArgs) -> None:
         Scenario(
             batteries_in_pool[1],
             {"capacity": 0},
-            Sample(now, Quantity(0.0)),
+            Sample(now, Percentage.from_percent(0.0)),
         ),
         Scenario(
             batteries_in_pool[0],
             {"capacity": 50, "soc": 55.0},
-            Sample(now, Quantity(50.0)),
+            Sample(now, Percentage.from_percent(50.0)),
         ),
         Scenario(
             batteries_in_pool[1],
             {"capacity": 150},
-            Sample(now, Quantity(25.0)),
+            Sample(now, Percentage.from_percent(25.0)),
         ),
     ]
 
@@ -757,15 +757,15 @@ async def run_soc_test(setup_args: SetupArgs) -> None:
         all_batteries=all_batteries,
         batteries_in_pool=batteries_in_pool,
         waiting_time_sec=waiting_time_sec,
-        all_pool_result=Sample(now, Quantity(25.0)),
-        only_first_battery_result=Sample(now, Quantity(50.0)),
+        all_pool_result=Sample(now, Percentage.from_percent(25.0)),
+        only_first_battery_result=Sample(now, Percentage.from_percent(50.0)),
     )
 
     # One battery stopped sending data.
     await streamer.stop_streaming(batteries_in_pool[1])
     await asyncio.sleep(MAX_BATTERY_DATA_AGE_SEC + 0.2)
     msg = await asyncio.wait_for(receiver.receive(), timeout=waiting_time_sec)
-    compare_messages(msg, Sample(now, Quantity(50.0)), 0.2)
+    compare_messages(msg, Sample(now, Percentage.from_percent(50.0)), 0.2)
 
     # All batteries stopped sending data.
     await streamer.stop_streaming(batteries_in_pool[0])
@@ -777,7 +777,7 @@ async def run_soc_test(setup_args: SetupArgs) -> None:
     latest_data = streamer.get_current_component_data(batteries_in_pool[0])
     streamer.start_streaming(latest_data, sampling_rate=0.1)
     msg = await asyncio.wait_for(receiver.receive(), timeout=waiting_time_sec)
-    compare_messages(msg, Sample(now, Quantity(50.0)), 0.2)
+    compare_messages(msg, Sample(now, Percentage.from_percent(50.0)), 0.2)
 
 
 async def run_power_bounds_test(  # pylint: disable=too-many-locals

tests/timeseries/test_quantities.py

@@ -10,6 +10,7 @@
 from frequenz.sdk.timeseries._quantities import (
     Current,
     Energy,
+    Percentage,
     Power,
     Quantity,
     Voltage,
@@ -257,3 +258,13 @@ def test_quantity_compositions() -> None:
     assert energy / power == timedelta(hours=6.2)
     assert energy / timedelta(hours=6.2) == power
     assert energy == power * timedelta(hours=6.2)
+
+
+def test_percentage() -> None:
+    """Test the percentage class."""
+    pct = Percentage.from_fraction(0.204)
+    assert f"{pct}" == "20.4 %"
+    pct = Percentage.from_percent(20.4)
+    assert f"{pct}" == "20.4 %"
+    assert pct.as_percent() == 20.4
+    assert pct.as_fraction() == 0.204