Clean ups and improvements to the resampler (mostly) (#68)

This PR mainly moves classes from the resampler to the new `timeseries`
package, but it also includes some nox and test improvements and fixes.

It also uses the timestamp produced by the timer as the resampling
timestamp, unifying it across all the resampling levels for full consistency.

- Allow passing positional arguments to nox sessions
- Move resampling actor to frequenz.sdk.actor
- Move resampler classes to frequenz.sdk.timeseries.resampler
- Remove unnecessary usage of pytz
- Fix sample ordering in tests
- Use more intuitive variable names in resampler tests
- Use easier to debug fake times in tests
- Allow the resampler to use an external timestamp
- Use the timestamp provided by the timer receiver
- Document ResamplingFunction
- Move Sample to frequenz.sdk.timeseries

Author: llucax

examples/sdk_resampling_example.py

@@ -13,10 +13,11 @@
 
 from frequenz.sdk.actor import ChannelRegistry
 from frequenz.sdk.actor.data_sourcing import DataSourcingActor
-from frequenz.sdk.data_ingestion.resampling.component_metrics_resampling_actor import (
+from frequenz.sdk.actor.resampling import (
+    ComponentMetricId,
+    ComponentMetricRequest,
     ComponentMetricsResamplingActor,
 )
-from frequenz.sdk.data_pipeline import ComponentMetricId, ComponentMetricRequest
 from frequenz.sdk.microgrid import ComponentCategory, microgrid_api
 
 HOST = "microgrid.sandbox.api.frequenz.io"

minimum-requirements-ci.txt

@@ -1,6 +1,6 @@
 # CI must ensure these dependency versions are supported
 frequenz-api-microgrid==0.11.0
-frequenz-channels==0.10.0
+frequenz-channels@git+https://github.com/frequenz-floss/[email protected]
 google-api-python-client==2.15
 grpcio==1.47
 grpcio-tools==1.47
@@ -9,7 +9,6 @@ pandas==1.3.5
 protobuf==3.20.2
 pyarrow==6.0.0
 pydantic==1.9.0
-pytz==2021.3
 sympy==1.10.1
 toml==0.10
 tqdm==4.38.0

noxfile.py

@@ -53,8 +53,9 @@ def formatting(session: nox.Session, install_deps: bool = True) -> None:
     if install_deps:
         session.install(*FMT_DEPS)
 
-    session.run("black", "--check", *source_file_paths(session))
-    session.run("isort", "--check", *source_file_paths(session))
+    paths = source_file_paths(session)
+    session.run("black", "--check", *paths)
+    session.run("isort", "--check", *paths)
 
 
 @nox.session
@@ -65,13 +66,6 @@ def mypy(session: nox.Session, install_deps: bool = True) -> None:
         # fast local tests with `nox -R -e mypy`.
         session.install("-e", ".", "mypy", *PYTEST_DEPS)
 
-    # Since we use other packages in the frequenz namespace, we need to run the
-    # checks for frequenz.sdk from the installed package instead of the src
-    # directory.
-    mypy_paths = [
-        path for path in source_file_paths(session) if not path.startswith("src")
-    ]
-
     mypy_cmd = [
         "mypy",
         "--ignore-missing-imports",
@@ -83,9 +77,21 @@ def mypy(session: nox.Session, install_deps: bool = True) -> None:
         "--strict",
     ]
 
+    # If we have session arguments, we just use those...
+    if session.posargs:
+        session.run(*mypy_cmd, *session.posargs)
+        return
+
     # Runs on the installed package
     session.run(*mypy_cmd, "-p", "frequenz.sdk")
+
     # Runs on the rest of the source folders
+    # Since we use other packages in the frequenz namespace, we need to run the
+    # checks for frequenz.sdk from the installed package instead of the src
+    # directory.
+    mypy_paths = [
+        path for path in source_file_paths(session) if not path.startswith("src")
+    ]
     session.run(*mypy_cmd, *mypy_paths)
 
 
@@ -97,10 +103,11 @@ def pylint(session: nox.Session, install_deps: bool = True) -> None:
         # fast local tests with `nox -R -e pylint`.
         session.install("-e", ".", "pylint", *PYTEST_DEPS)
 
+    paths = source_file_paths(session)
     session.run(
         "pylint",
         "--extension-pkg-whitelist=pydantic",
-        *source_file_paths(session),
+        *paths,
     )
 
 
@@ -110,12 +117,14 @@ def docstrings(session: nox.Session, install_deps: bool = True) -> None:
     if install_deps:
         session.install(*DOCSTRING_DEPS, "toml")
 
-    session.run("pydocstyle", *source_file_paths(session))
+    paths = source_file_paths(session)
+    session.run("pydocstyle", *paths)
 
     # Darglint checks that function argument and return values are documented.
     # This is needed only for the `src` dir, so we exclude the other top level
-    # dirs that contain code.
-    darglint_paths = filter(
+    # dirs that contain code, unless some paths were specified by argument, in
+    # which case we use those untouched.
+    darglint_paths = session.posargs or filter(
         lambda path: not (path.startswith("tests") or path.startswith("benchmarks")),
         source_file_paths(session),
     )
@@ -158,4 +167,5 @@ def _pytest_impl(
         "--cov=frequenz.sdk",
         "--cov-report=term",
         f"--cov-report=html:.htmlcov-{max_or_min_deps}",
+        *session.posargs,
     )

pyproject.toml

@@ -27,7 +27,7 @@ classifiers = [
 requires-python = ">= 3.8, < 4"
 dependencies = [
     "frequenz-api-microgrid >= 0.11.0, < 0.12.0",
-    "frequenz-channels >= 0.10.0, < 0.11.0",
+    "frequenz-channels @ git+https://github.com/frequenz-floss/[email protected]",
     "google-api-python-client >= 2.15, < 3",
     "grpcio >= 1.47, < 2",
     "grpcio-tools >= 1.47, < 2",
@@ -36,7 +36,6 @@ dependencies = [
     "protobuf >= 3.20, < 4",
     "pyarrow >= 6.0.0, < 6.1",
     "pydantic >= 1.9",
-    "pytz >= 2021.3",
     "sympy >= 1.10.1, < 2",
     "toml >= 0.10",
     "tqdm >= 4.38.0, < 5",

src/frequenz/sdk/actor/data_sourcing/__init__.py

@@ -9,7 +9,10 @@
 """
 
 from .data_sourcing import DataSourcingActor
+from .types import ComponentMetricId, ComponentMetricRequest
 
 __all__ = [
     "DataSourcingActor",
+    "ComponentMetricId",
+    "ComponentMetricRequest",
 ]

src/frequenz/sdk/actor/data_sourcing/data_sourcing.py

@@ -9,11 +9,9 @@
 
 from frequenz.channels import Receiver
 
-from frequenz.sdk.actor import actor
-from frequenz.sdk.actor.channel_registry import ChannelRegistry
-from frequenz.sdk.data_pipeline import ComponentMetricRequest
-
+from .. import ChannelRegistry, actor
 from .microgrid_api_source import MicrogridApiSource
+from .types import ComponentMetricRequest
 
 
 @actor

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

@@ -13,16 +13,17 @@
 
 from frequenz.channels import Receiver, Sender
 
-from frequenz.sdk.actor import ChannelRegistry
-from frequenz.sdk.data_pipeline import ComponentMetricId, ComponentMetricRequest, Sample
-from frequenz.sdk.microgrid import (
+from ...microgrid import (
     BatteryData,
     ComponentCategory,
     EVChargerData,
     InverterData,
     MeterData,
     microgrid_api,
 )
+from ...timeseries import Sample
+from .. import ChannelRegistry
+from .types import ComponentMetricId, ComponentMetricRequest
 
 _MeterDataMethods: Dict[ComponentMetricId, Callable[[MeterData], float]] = {
     ComponentMetricId.ACTIVE_POWER: lambda msg: msg.active_power,

src/frequenz/sdk/data_pipeline/types.py …requenz/sdk/actor/data_sourcing/types.pysrc/frequenz/sdk/data_pipeline/types.py renamed to src/frequenz/sdk/actor/data_sourcing/types.py src/frequenz/sdk/data_pipeline/types.py renamed to src/frequenz/sdk/actor/data_sourcing/types.py

@@ -73,16 +73,3 @@ def get_channel_name(self) -> str:
             A string denoting a channel name.
         """
         return f"{self.component_id}::{self.metric_id.name}::{self.start_time}::{self.namespace}"
-
-
-@dataclass(frozen=True)
-class Sample:
-    """A measurement taken at a particular point in time.
-
-    The `value` could be `None` if a component is malfunctioning or data is
-    lacking for another reason, but a sample still needs to be sent to have a
-    coherent view on a group of component metrics for a particular timestamp.
-    """
-
-    timestamp: datetime
-    value: Optional[float] = None

…ng/component_metrics_resampling_actor.py src/frequenz/sdk/actor/resampling.pysrc/frequenz/sdk/data_ingestion/resampling/component_metrics_resampling_actor.py renamed to src/frequenz/sdk/actor/resampling.py src/frequenz/sdk/data_ingestion/resampling/component_metrics_resampling_actor.py renamed to src/frequenz/sdk/actor/resampling.py

@@ -15,14 +15,19 @@
 
 from frequenz.channels import MergeNamed, Receiver, Select, Sender, Timer
 
-from ...actor import ChannelRegistry, actor
-from ...data_pipeline import ComponentMetricRequest, Sample
-from .component_metric_group_resampler import ComponentMetricGroupResampler
-from .component_metric_resampler import ResamplingFunction
+from ..timeseries import GroupResampler, ResamplingFunction, Sample
+from . import ChannelRegistry, actor, data_sourcing
 
 logger = logging.Logger(__name__)
 
 
+# Re-export the types from the data_sourcing actor as we use the same requests,
+# we are only forwarding them for now.
+ComponentMetricId = data_sourcing.ComponentMetricId
+
+ComponentMetricRequest = data_sourcing.ComponentMetricRequest
+
+
 # pylint: disable=unused-argument
 def average(samples: Sequence[Sample], resampling_period_s: float) -> float:
     """Calculate average of the provided values.
@@ -147,7 +152,7 @@ async def run() -> None:
         self._max_data_age_in_periods: float = max_data_age_in_periods
         self._resampling_function: ResamplingFunction = resampling_function
 
-        self._resampler = ComponentMetricGroupResampler(
+        self._resampler = GroupResampler(
             resampling_period_s=resampling_period_s,
             max_data_age_in_periods=max_data_age_in_periods,
             initial_resampling_function=resampling_function,
@@ -209,10 +214,12 @@ async def run(self) -> None:
                 component_data_receiver=MergeNamed(**self._input_receivers),
             )
             while await select.ready():
-                if _ := select.resampling_timer:
+                if msg := select.resampling_timer:
+                    assert msg.inner is not None, "The timer should never be 'closed'"
+                    timestamp = msg.inner
                     awaitables = [
                         self._output_senders[channel_name].send(sample)
-                        for channel_name, sample in self._resampler.resample()
+                        for channel_name, sample in self._resampler.resample(timestamp)
                     ]
                     await asyncio.gather(*awaitables)
                 if msg := select.component_data_receiver:

src/frequenz/sdk/data_handling/time_series.py

@@ -9,14 +9,12 @@
 """
 from __future__ import annotations
 
-import datetime
 import enum
 import logging
 from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
 from typing import Collection, Dict, Generic, Optional, Set, TypeVar
 
-import pytz
-
 from .formula import Formula
 
 Key = TypeVar("Key")
@@ -104,13 +102,13 @@ class Status(enum.Enum):
         UNKNOWN = "unknown"
         ERROR = "error"
 
-    timestamp: datetime.datetime
+    timestamp: datetime
     value: Optional[Value] = None
     status: Status = Status.VALID
     broken_component_ids: Set[int] = field(default_factory=set)
 
     @staticmethod
-    def create_error(timestamp: datetime.datetime) -> TimeSeriesEntry[Value]:
+    def create_error(timestamp: datetime) -> TimeSeriesEntry[Value]:
         """Create a `TimeSeriesEntry` that contains an error.
 
         This can happen when the value would be NaN, e.g.
@@ -128,7 +126,7 @@ def create_error(timestamp: datetime.datetime) -> TimeSeriesEntry[Value]:
 
     @staticmethod
     def create_unknown(
-        timestamp: datetime.datetime, broken_component_ids: Optional[Set[int]] = None
+        timestamp: datetime, broken_component_ids: Optional[Set[int]] = None
     ) -> TimeSeriesEntry[Value]:
         """Create a `TimeSeriesEntry` that contains an unknown value.
 
@@ -174,11 +172,11 @@ class LatestEntryCache(Generic[Key, Value]):
 
     def __init__(self) -> None:
         """Initialize the class."""
-        self._latest_timestamp = pytz.utc.localize(datetime.datetime.min)
+        self._latest_timestamp = datetime.min.replace(tzinfo=timezone.utc)
         self._entries: Dict[Key, TimeSeriesEntry[Value]] = {}
 
     @property
-    def latest_timestamp(self) -> datetime.datetime:
+    def latest_timestamp(self) -> datetime:
         """Get the most recently observed timestamp across all keys in the cache.
 
         Returns:
@@ -209,7 +207,7 @@ def __contains__(self, key: Key) -> bool:
     def get(
         self,
         key: Key,
-        timedelta_tolerance: datetime.timedelta = datetime.timedelta.max,
+        timedelta_tolerance: timedelta = timedelta.max,
         default: Optional[TimeSeriesEntry[Value]] = None,
     ) -> CacheEntryLookupResult[Value]:
         """Get the cached entry for the specified key, if any.
@@ -232,7 +230,7 @@ def get(
                 retrieved from the cache has a timestamp greater than the latest saved
                 timestamp across all cache keys.
         """
-        if timedelta_tolerance < datetime.timedelta(0):
+        if timedelta_tolerance < timedelta(0):
             raise ValueError(
                 f"timedelta_tolerance cannot be less than 0, but "
                 f"{timedelta_tolerance} was provided"
@@ -320,7 +318,7 @@ def reset(self) -> None:
         slightly more efficient.
         """
         self.clear()
-        self._latest_timestamp = pytz.utc.localize(datetime.datetime.min)
+        self._latest_timestamp = datetime.min.replace(tzinfo=timezone.utc)
 
     def reset_latest_timestamp(self) -> bool:
         """Reset the `latest_timestamp` property to the lowest possible value.
@@ -342,7 +340,7 @@ def reset_latest_timestamp(self) -> bool:
 
         self._latest_timestamp = max(
             map(lambda x: x.timestamp, self._entries.values()),
-            default=pytz.utc.localize(datetime.datetime.min),
+            default=datetime.min.replace(tzinfo=timezone.utc),
         )
 
         if self._latest_timestamp > previous:
@@ -423,7 +421,7 @@ def evaluate(
         cache: LatestEntryCache[str, Value],
         formula_name: str = "",
         symbol_to_symbol_mapping: Optional[Dict[str, SymbolMapping]] = None,
-        timedelta_tolerance: datetime.timedelta = datetime.timedelta.max,
+        timedelta_tolerance: timedelta = timedelta.max,
         default_entry: Optional[TimeSeriesEntry[Value]] = None,
     ) -> Optional[TimeSeriesEntry[Value]]:
         """Evaluate the formula using time-series values from the provided cache.
@@ -463,7 +461,7 @@ def evaluate(
                 `timedelta_tolerance`
         """
         kwargs: Dict[str, Optional[Value]] = {}
-        timestamp = pytz.utc.localize(datetime.datetime.min)
+        timestamp = datetime.min.replace(tzinfo=timezone.utc)
 
         symbol_to_symbol_mapping = symbol_to_symbol_mapping or {}
         formula_broken_component_ids: Set[int] = set()