Single value processing of signals (#37)

* single value data

* fix test

* single value data

* fix test

---------

Co-authored-by: Frances Hartwell <[email protected]>

Author: sarahmish

DEVELOPMENT.md

@@ -259,11 +259,12 @@ imported and used like this:
 from sigpro.demo import get_amplitude_demo
 from sigpro.demo import get_frequency_demo
 from sigpro.demo import get_frequency_time_demo
+from sigpro.demo import get_demo_data 
 
 amplitude_values, sampling_frequency = get_amplitude_demo()
 amplitude_values, frequency_values = get_frequency_demo()
 amplitude_values, frequency_values, time_values = get_frequency_time_demo()
-dataframe = get_frequency_time_demo(dataframe=True)
+dataframe = get_demo_data()
 ```
 
 In all cases, the functions will return values that correspond to a

sigpro/core.py

@@ -163,19 +163,28 @@ def __init__(self, transformations, aggregations, values_column_name='values',
         self.input_is_dataframe = input_is_dataframe
         self.pipeline = self._build_pipeline()
 
-    def _apply_pipeline(self, row):
+    def _apply_pipeline(self, window, is_series=False):
         """Apply a ``mlblocks.MLPipeline`` to a row.
 
-        Apply a ``MLPipeline`` to a row of a ``pd.DataFrame``, this function can
+        Apply a ``MLPipeline`` to a window of a ``pd.DataFrame``, this function can
         be combined with the ``pd.DataFrame.apply`` method to be applied to the
         entire data frame.
 
         Args:
-            row (pd.Series):
-                Row used to apply the pipeline to.
+            window (pd.Series):
+                Row or multiple rows (window) used to apply the pipeline to.
+            is_series (bool):
+                Indicator whether window is formated as a series or dataframe.
         """
-        context = row.to_dict()
-        amplitude_values = context.pop(self.values_column_name)
+        if is_series:
+            context = window.to_dict()
+            amplitude_values = context.pop(self.values_column_name)
+        else:
+            context = {} if window.empty else {
+                k: v for k, v in window.iloc[0].to_dict().items() if k != self.values_column_name
+            }
+            amplitude_values = list(window[self.values_column_name])
+
         output = self.pipeline.predict(
             amplitude_values=amplitude_values,
             **context,
@@ -187,12 +196,19 @@ def _apply_pipeline(self, row):
 
         return pd.Series(dict(zip(output_names, output)))
 
-    def process_signal(self, data=None, feature_columns=None, **kwargs):
+    def process_signal(self, data=None, window=None, time_index=None, groupby_index=None,
+                       feature_columns=None, **kwargs):
         """Apply multiple transformation and aggregation primitives.
 
         Args:
             data (pandas.DataFrame):
                 Dataframe with a column that contains signal values.
+            window (str):
+                Duration of window size, e.g. ('1h').
+            time_index (str):
+                Column in ``data`` that represents the time index.
+            groupby_index (str or list[str]):
+                Column(s) to group together and take the window over.
             feature_columns (list):
                 List of column names from the input data frame that must be considered as
                 features and should not be dropped.
@@ -207,15 +223,25 @@ def process_signal(self, data=None, feature_columns=None, **kwargs):
                     A list with the feature names generated.
         """
         if data is None:
-            row = pd.Series(kwargs)
-            values = self._apply_pipeline(row).values
+            window = pd.Series(kwargs)
+            values = self._apply_pipeline(window, is_series=True).values
             return values if len(values) > 1 else values[0]
 
-        features = data.apply(
-            self._apply_pipeline,
-            axis=1
-        )
-        data = pd.concat([data, features], axis=1)
+        data = data.copy()
+        if window is not None and groupby_index is not None:
+            features = data.set_index(time_index).groupby(groupby_index).resample(
+                rule=window, **kwargs).apply(
+                self._apply_pipeline
+            ).reset_index()
+            data = features
+
+        else:
+            features = data.apply(
+                self._apply_pipeline,
+                axis=1,
+                is_series=True
+            )
+            data = pd.concat([data, features], axis=1)
 
         if feature_columns:
             feature_columns = feature_columns + list(features.columns)

sigpro/demo.py

@@ -11,16 +11,30 @@
 DEMO_PATH = os.path.join(os.path.dirname(__file__), 'data')
 
 
+def _load_demo(nrows=None):
+    demo_path = os.path.join(DEMO_PATH, 'demo_timeseries.csv')
+    df = pd.read_csv(demo_path, parse_dates=['timestamp'], nrows=nrows)
+    df['sampling_frequency'] = 1000
+    df["values"] = df["values"].apply(json.loads).apply(list)
+
+    return df
+
+
 def get_demo_data(nrows=None):
     """Get a demo ``pandas.DataFrame`` containing the accepted data format.
 
+    Args:
+        nrows (int):
+            Number of rows to load from the demo datasets.
+
     Returns:
         A ``pd.DataFrame`` containing as ``values`` the signal values.
     """
-    demo_path = os.path.join(DEMO_PATH, 'demo_timeseries.csv')
-    df = pd.read_csv(demo_path, parse_dates=['timestamp'], nrows=nrows)
-    df['sampling_frequency'] = 1000
-    df["values"] = df["values"].apply(json.loads).apply(list)
+    df = _load_demo(nrows)
+    df = df.explode('values').reset_index(drop=True)
+
+    time_delta = pd.to_timedelta(list(range(400)) * 750, 's')
+    df['timestamp'] = df['timestamp'] + time_delta
     return df
 
 
@@ -71,7 +85,7 @@ def get_amplitude_demo(index=None):
             A tuple with a `np.array` containing amplitude values and as second element the
             sampling frequency used.
     """
-    df = get_demo_data()
+    df = _load_demo()
     if index is None:
         index = random.randint(0, len(df))
 

tests/integration/test_demo.py

@@ -5,7 +5,7 @@
     get_amplitude_demo, get_demo_data, get_demo_primitives, get_frequency_demo,
     get_frequency_time_demo)
 
-EXPECTED_SHAPE = (750, 5)
+EXPECTED_SHAPE = (300000, 5)
 EXPECTED_COLUMNS = ['turbine_id', 'signal_id', 'timestamp', 'values', 'sampling_frequency']
 EXPECTED_SAMPLING_FREQUENCY = 10000
 EXPECTED_VALUES_LENGTH = 400
@@ -54,7 +54,7 @@ def test_get_frequency_demo_indexed():
 def test_get_frequency_demo_complex():
     values, frequency_values = get_frequency_demo(real=False)
     value = values[0]
-    assert type(value) == np.complex128
+    assert isinstance(value, np.complex128)
     assert EXPECTED_VALUES_LENGTH // 2 == len(values)
     assert EXPECTED_FREQUENCY_LENGTH // 2 == len(frequency_values)
 
@@ -79,4 +79,4 @@ def test_get_frequency_time_demo_complex():
     assert 129 == len(values)
     assert 129 == len(frequencies)
     assert 5 == len(time_values)
-    assert type(value) == np.complex128
+    assert isinstance(value, np.complex128)