Move Sample to frequenz.sdk.timeseries

Now that we have a timeseries package it makes more sense to have the
Sample class living there.

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

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

@@ -14,7 +14,7 @@
 from frequenz.channels import Receiver, Sender
 
 from frequenz.sdk.actor import ChannelRegistry
-from frequenz.sdk.data_pipeline import ComponentMetricId, ComponentMetricRequest, Sample
+from frequenz.sdk.data_pipeline import ComponentMetricId, ComponentMetricRequest
 from frequenz.sdk.microgrid import (
     BatteryData,
     ComponentCategory,
@@ -23,6 +23,7 @@
     MeterData,
     microgrid_api,
 )
+from frequenz.sdk.timeseries import Sample
 
 _MeterDataMethods: Dict[ComponentMetricId, Callable[[MeterData], float]] = {
     ComponentMetricId.ACTIVE_POWER: lambda msg: msg.active_power,

src/frequenz/sdk/actor/resampling.py

@@ -15,7 +15,8 @@
 
 from frequenz.channels import MergeNamed, Receiver, Select, Sender, Timer
 
-from ..data_pipeline import ComponentMetricRequest, Sample
+from ..data_pipeline import ComponentMetricRequest
+from ..timeseries import Sample
 from ..timeseries.resampler import ComponentMetricGroupResampler, ResamplingFunction
 from . import ChannelRegistry, actor
 

src/frequenz/sdk/data_pipeline/__init__.py

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

src/frequenz/sdk/data_pipeline/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

src/frequenz/sdk/timeseries/__init__.py

@@ -18,9 +18,11 @@
     ComponentMetricResampler,
     ResamplingFunction,
 )
+from .sample import Sample
 
 __all__ = [
     "ComponentMetricGroupResampler",
     "ComponentMetricResampler",
     "ResamplingFunction",
+    "Sample",
 ]

src/frequenz/sdk/timeseries/resampler.py

@@ -13,7 +13,7 @@
 from datetime import datetime, timedelta, timezone
 from typing import Callable, Deque, Dict, Generator, Optional, Sequence, Tuple
 
-from ..data_pipeline import Sample
+from .sample import Sample
 
 logger = logging.Logger(__name__)
 

src/frequenz/sdk/timeseries/sample.py

@@ -0,0 +1,25 @@
+"""Common types for the Data Pipeline.
+
+Copyright
+Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+License
+MIT
+"""
+
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Optional
+
+
+@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

tests/timeseries/test_resampling.py

@@ -13,10 +13,10 @@
 
 import time_machine
 
-from frequenz.sdk.data_pipeline import Sample
-from frequenz.sdk.timeseries.resampler import (
+from frequenz.sdk.timeseries import (
     ComponentMetricGroupResampler,
     ComponentMetricResampler,
+    Sample,
 )