Improve resampler structure and performance (#1242)

This pull request splits the `resampling` module into many submodules as
it was getting very large

If also uses tuples internally to store and calculate samples, which
improves performance and simplifies the logic, as `Sample`s can have
`None` values, but the values in the internal buffer can never be
`None`, and we don't need to use a `Quantity` as we know we are always
working with the same units.

It also exposes the `ResamplingFunction` and `SourceProperties` types,
converts `ResamplingFunction` to a `Protocol`, and replaces `average` as
the default resampling function with Python's `statistics.fmean`, among
other small improvements and fixes.

Author: llucax

RELEASE_NOTES.md

@@ -7,6 +7,8 @@
 ## Upgrading
 
 * The microgrid client dependency has been updated to version 0.9.0
+* The resampling function now takes plain `float`s as values instead of `Quantity` objects.
+* `frequenz.sdk.timeseries.UNIX_EPOCH` was removed, use [`frequenz.core.datetime.UNIX_EPOCH`](https://frequenz-floss.github.io/frequenz-core-python/latest/reference/frequenz/core/datetime/#frequenz.core.datetime.UNIX_EPOCH) instead.
 
 ## New Features
 

benchmarks/timeseries/resampling.py

@@ -7,18 +7,13 @@
 from datetime import datetime, timedelta, timezone
 from timeit import timeit
 
-from frequenz.quantities import Quantity
-
-from frequenz.sdk.timeseries import Sample
-from frequenz.sdk.timeseries._resampling import (
-    ResamplerConfig,
-    SourceProperties,
-    _ResamplingHelper,
-)
+from frequenz.sdk.timeseries import ResamplerConfig
+from frequenz.sdk.timeseries._resampling._base_types import SourceProperties
+from frequenz.sdk.timeseries._resampling._resampler import _ResamplingHelper
 
 
 def nop(  # pylint: disable=unused-argument
-    samples: Sequence[Sample[Quantity]],
+    samples: Sequence[tuple[datetime, float]],
     resampler_config: ResamplerConfig,
     source_properties: SourceProperties,
 ) -> float:
@@ -43,10 +38,11 @@ def _benchmark_resampling_helper(resamples: int, samples: int) -> None:
 
     def _do_work() -> None:
         nonlocal now
+        delta = timedelta(seconds=1 / samples)
         for _n_resample in range(resamples):
             for _n_sample in range(samples):
-                now = now + timedelta(seconds=1 / samples)
-                helper.add_sample(Sample(now, Quantity(0.0)))
+                now = now + delta
+                helper.add_sample((now, 0.0))
             helper.resample(now)
 
     print(timeit(_do_work, number=5))

examples/battery_pool.py

@@ -33,9 +33,8 @@ async def main() -> None:
     receivers = [
         battery_pool.soc.new_receiver(limit=1),
         battery_pool.capacity.new_receiver(limit=1),
-        # pylint: disable=protected-access
+        # pylint: disable-next=protected-access
         battery_pool._system_power_bounds.new_receiver(limit=1),
-        # pylint: enable=protected-access
     ]
 
     async for metric in merge(*receivers):

mkdocs.yml

@@ -117,8 +117,10 @@ plugins:
           import:
             # See https://mkdocstrings.github.io/python/usage/#import for details
             - https://docs.python.org/3/objects.inv
-            - https://frequenz-floss.github.io/frequenz-channels-python/v1.1/objects.inv
-            - https://frequenz-floss.github.io/frequenz-client-microgrid-python/v0.7/objects.inv
+            - https://frequenz-floss.github.io/frequenz-channels-python/v1/objects.inv
+            - https://frequenz-floss.github.io/frequenz-client-common-python/v0.3/objects.inv
+            - https://frequenz-floss.github.io/frequenz-client-microgrid-python/v0.9/objects.inv
+            - https://frequenz-floss.github.io/frequenz-core-python/v1/objects.inv
             - https://frequenz-floss.github.io/frequenz-quantities-python/v1/objects.inv
             - https://lovasoa.github.io/marshmallow_dataclass/html/objects.inv
             - https://marshmallow.readthedocs.io/en/stable/objects.inv

src/frequenz/sdk/actor/__init__.py

@@ -597,7 +597,7 @@ async def main() -> None:  # (6)!
 [_run]: #the-_run-method
 """
 
-from ..timeseries._resampling import ResamplerConfig
+from ..timeseries._resampling._config import ResamplerConfig
 from ._actor import Actor
 from ._background_service import BackgroundService
 from ._run_utils import run

src/frequenz/sdk/microgrid/_power_distributing/_component_status/_battery_status_tracker.py

@@ -394,8 +394,8 @@ def _is_inverter_state_correct(self, msg: InverterData) -> bool:
             True if inverter is in correct state. False otherwise.
         """
         # Component state is not exposed to the user.
-        # pylint: disable=protected-access
         state = msg.component_state
+        # pylint: disable-next=protected-access
         if state not in BatteryStatusTracker._inverter_valid_state:
             if self._last_status == ComponentStatusEnum.WORKING:
                 _logger.warning(
@@ -404,7 +404,6 @@ def _is_inverter_state_correct(self, msg: InverterData) -> bool:
                     state.name,
                 )
             return False
-        # pylint: enable=protected-access
         return True
 
     def _is_battery_state_correct(self, msg: BatteryData) -> bool:
@@ -417,8 +416,8 @@ def _is_battery_state_correct(self, msg: BatteryData) -> bool:
             True if battery is in correct state. False otherwise.
         """
         # Component state is not exposed to the user.
-        # pylint: disable=protected-access
         state = msg.component_state
+        # pylint: disable-next=protected-access
         if state not in BatteryStatusTracker._battery_valid_state:
             if self._last_status == ComponentStatusEnum.WORKING:
                 _logger.warning(
@@ -439,7 +438,6 @@ def _is_battery_state_correct(self, msg: BatteryData) -> bool:
                 )
             return False
         return True
-        # pylint: enable=protected-access
 
     def _is_timestamp_outdated(self, timestamp: datetime) -> bool:
         """Return if timestamp is to old.

src/frequenz/sdk/microgrid/_power_distributing/_distribution_algorithm/_battery_distribution_algorithm.py

@@ -461,7 +461,8 @@ def _compute_battery_availability_ratio(
 
         return battery_availability_ratio, total_battery_availability_ratio
 
-    def _distribute_power(  # pylint: disable=too-many-arguments
+    # pylint: disable-next=too-many-arguments,too-many-locals,too-many-branches,too-many-statements
+    def _distribute_power(
         self,
         *,
         components: list[InvBatPair],
@@ -470,7 +471,6 @@ def _distribute_power(  # pylint: disable=too-many-arguments
         incl_bounds: dict[ComponentId, Power],
         excl_bounds: dict[ComponentId, Power],
     ) -> DistributionResult:
-        # pylint: disable=too-many-locals,too-many-branches,too-many-statements
         """Distribute power between given components.
 
         After this method power should be distributed between batteries

src/frequenz/sdk/microgrid/_power_distributing/power_distributing.py

@@ -33,8 +33,7 @@
 _logger = logging.getLogger(__name__)
 
 
-class PowerDistributingActor(Actor):
-    # pylint: disable=too-many-instance-attributes
+class PowerDistributingActor(Actor):  # pylint: disable=too-many-instance-attributes
     """Actor to distribute the power between components in a microgrid.
 
     One instance of the actor can handle only one component category and type,

src/frequenz/sdk/microgrid/_power_managing/_power_managing_actor.py

@@ -158,16 +158,17 @@ def _add_system_bounds_tracker(self, component_ids: frozenset[ComponentId]) -> N
             NotImplementedError: When the pool type is not supported.
         """
         bounds_receiver: Receiver[SystemBounds]
-        # pylint: disable=protected-access
         if self._component_category is ComponentCategory.BATTERY:
             battery_pool = _data_pipeline.new_battery_pool(
                 priority=-sys.maxsize - 1, component_ids=component_ids
             )
+            # pylint: disable-next=protected-access
             bounds_receiver = battery_pool._system_power_bounds.new_receiver()
         elif self._component_category is ComponentCategory.EV_CHARGER:
             ev_charger_pool = _data_pipeline.new_ev_charger_pool(
                 priority=-sys.maxsize - 1, component_ids=component_ids
             )
+            # pylint: disable-next=protected-access
             bounds_receiver = ev_charger_pool._system_power_bounds.new_receiver()
         elif (
             self._component_category is ComponentCategory.INVERTER
@@ -176,8 +177,8 @@ def _add_system_bounds_tracker(self, component_ids: frozenset[ComponentId]) -> N
             pv_pool = _data_pipeline.new_pv_pool(
                 priority=-sys.maxsize - 1, component_ids=component_ids
             )
+            # pylint: disable-next=protected-access
             bounds_receiver = pv_pool._system_power_bounds.new_receiver()
-        # pylint: enable=protected-access
         else:
             err = (
                 "PowerManagingActor: Unsupported component category: "

src/frequenz/sdk/microgrid/_power_wrapper.py

@@ -9,14 +9,16 @@
 from datetime import timedelta
 
 from frequenz.channels import Broadcast
-
-# pylint seems to think this is a cyclic import, but it is not.
-#
-# pylint: disable=cyclic-import
 from frequenz.client.microgrid import ComponentCategory, ComponentType
 
 from .._internal._channels import ChannelRegistry, ReceiverFetcher
+
+# pylint seems to think this is a cyclic import, but it is not.
+#
+# pylint: disable-next=cyclic-import
 from . import _power_managing, connection_manager
+
+# pylint: disable-next=cyclic-import
 from ._power_distributing import (
     ComponentPoolStatus,
     PowerDistributingActor,