Repository: KIT-IAI/pyWATTS

Issue:
Performance problems in online learning
**Describe the bug**
Online learning is currently slow - taking large amounts of time to process one step. It seems that a single processor is busy. This could be due to:
- Merging and copying of xarray.datasets.
- Predict/Transform of single datapoints by sklearn.
- Internal calls of pyWATTS

**Expected behavior**
The online learning should run smoothly. Posisble solutions:
- Reduce copying and merging of datasets.
- Multiprocessing.


Discussion:

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index e52e5da..73f8604 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -35,6 +35,13 @@
 * Fix save and load of keras Wrapper  ([#91](https://github.com/KIT-IAI/pyWATTS/issues/91))
 * Import of pytorchWrapper via wrapper.__init__
 * Fix the mask for weekends and public holidays in rolling_base  ([#107](https://github.com/KIT-IAI/pyWATTS/issues/107))
+* Improved Online Learning Performance  ([#18](https://github.com/KIT-IAI/pyWATTS/issues/18))
+  * Add a cache for the most recent value, since this value is often requested by successing modules.
+  * Remove unecessary copies
+  * Avoid uncessary execution in should_stop 
+  * Improve performance in sample_module
+  * Improve performance in trend_extraction
+  * Callbacks are only executed if the step is finished. No intermediate results are plotted.
 * Fix the extracted time index in rolling RMSE  ([#124](https://github.com/KIT-IAI/pyWATTS/issues/124))
 * Use raise from if an exception is raise because an other is raised before for retaining the original stack trace  ([#107](https://github.com/KIT-IAI/pyWATTS/issues/123))
 
diff --git a/pywatts/core/base_step.py b/pywatts/core/base_step.py
index 4a09c38..e35be31 100644
--- a/pywatts/core/base_step.py
+++ b/pywatts/core/base_step.py
@@ -1,6 +1,7 @@
 import logging
 from abc import ABC, abstractmethod
 from typing import Optional, Dict
+import copy
 
 import pandas as pd
 import xarray as xr
@@ -25,7 +26,7 @@ class BaseStep(ABC):
     :type computation_mode: ComputationMode
     """
 
-    def __init__(self,  input_steps: Optional[Dict[str, "BaseStep"]]=None,
+    def __init__(self, input_steps: Optional[Dict[str, "BaseStep"]] = None,
                  targets: Optional[Dict[str, "BaseStep"]] = None, condition=None,
                  computation_mode=ComputationMode.Default):
         self._original_compuation_mode = computation_mode
@@ -33,6 +34,7 @@ class BaseStep(ABC):
         self.input_steps: Dict[str, "BaseStep"] = dict() if input_steps is None else input_steps
         self.targets: Dict[str, "BaseStep"] = dict() if targets is None else targets
         self.condition = condition
+        self.cached_result = {"cached": None, "start": None, "end": None}
 
         self.name = "BaseStep"
 
@@ -62,25 +64,35 @@ class BaseStep(ABC):
         :type return_all: bool
         :return: The resulting data or None if no data are calculated
         """
-
         # Check if step should be executed.
         if self._should_stop(start, end):
             return None
 
-        # Trigger fit and transform if necessary
-        if not self.finished:
+        # Only execute the module if the step is not finished and the results are not yet calculated
+        if not self.finished and not (end is not None and self._current_end is not None and end <= self._current_end):
             if not self.buffer or not self._current_end or end > self._current_end:
-                self._compute(start, end)
+                self.cached_result["cached"] = self._compute(start, end)
+                self.cached_result["start"] = start
+                self.cached_result["end"] = end
                 self._current_end = end
             if not end:
                 self.finished = True
             else:
                 self.finished = not self.further_elements(end)
-            self._callbacks()
 
+            # Only call callbacks if the step is finished
+            if self.finished:
+                self._callbacks()
+
+        # Check if the cached results fits to the request, if yes return it.
+        if self.cached_result["cached"] is not None and self.cached_result["start"] == start and self.cached_result[
+            "end"] == end:
+            return copy.deepcopy(self.cached_result["cached"]) if return_all else copy.deepcopy(self.cached_result["cached"][
+                buffer_element]) if buffer_element is not None else copy.deepcopy(list(self.cached_result["cached"].values())[
+                0])
         return self._pack_data(start, end, buffer_element, return_all=return_all)
 
-    def _compute(self, start, end):
+    def _compute(self, start, end) -> Dict[str, xr.DataArray]:
         pass
 
     def further_elements(self, counter: pd.Timestamp) -> bool:
@@ -109,23 +121,24 @@ class BaseStep(ABC):
         if end and start and end > start:
             index = list(self.buffer.values())[0].indexes[time_index[0]]
             start = max(index[0], start.to_numpy())
+            # After sel copy is not needed, since it returns a new array.
             if buffer_element is not None:
-                return self.buffer.copy()[buffer_element].sel(
+                return self.buffer[buffer_element].sel(
                     **{time_index[0]: index[(index >= start) & (index < end.to_numpy())]})
             elif return_all:
-                return {key: b.copy().sel(**{time_index[0]: index[(index >= start) & (index < end.to_numpy())]}) for
+                return {key: b.sel(**{time_index[0]: index[(index >= start) & (index < end.to_numpy())]}) for
                         key, b in self.buffer.items()}
             else:
-                return list(self.buffer.copy().values())[0].sel(
+                return list(self.buffer.values())[0].sel(
                     **{time_index[0]: index[(index >= start) & (index < end.to_numpy())]})
         else:
             self.finished = True
             if buffer_element is not None:
                 return self.buffer[buffer_element].copy()
             elif return_all:
-                return self.buffer.copy()
+                return copy.deepcopy(self.buffer)
             else:
-                return list(self.buffer.copy().values())[0]
+                return list(self.buffer.values())[0].copy()
 
     def _transform(self, input_step):
         pass
@@ -149,6 +162,7 @@ class BaseStep(ABC):
             dim = _get_time_indeces(result)[0]
             for key in self.buffer.keys():
                 self.buffer[key] = xr.concat([self.buffer[key], result[key]], dim=dim)
+        return result
 
     def get_json(self, fm: FileManager) -> Dict:
         """
@@ -192,13 +206,16 @@ class BaseStep(ABC):
 
     def _should_stop(self, start, end) -> bool:
         # Fetch input and target data
-        input_step = self._get_input(start, end)
-        target_step = self._get_target(start, end)
+        input_result = self._get_input(start, end)
+        target_result = self._get_target(start, end)
+
+        # Check if either the condition is True or some of the previous steps stopped (return_value is None)
+        return (self.condition is not None and not self.condition(input_result, target_result)) or \
+               self._input_stopped(input_result) or self._input_stopped(target_result)
 
-        return (self.condition is not None and not self.condition(input_step, target_step)) or \
-               (len(self.input_steps) > 0 and
-                any(map(lambda x: x._should_stop(start, end), self.input_steps.values()))) or \
-               (len(self.targets) > 0 and any(map(lambda x: x._should_stop(start, end), self.targets.values())))
+    @staticmethod
+    def _input_stopped(input_data):
+        return (input_data is not None and len(input_data) > 0 and any(map(lambda x: x is None, input_data.values())))
 
     def reset(self):
         """
diff --git a/pywatts/core/either_or_step.py b/pywatts/core/either_or_step.py
index 821965d..c7e0c6e 100644
--- a/pywatts/core/either_or_step.py
+++ b/pywatts/core/either_or_step.py
@@ -30,8 +30,7 @@ class EitherOrStep(BaseStep):
         for in_step in input_step:
             if in_step is not None:
                 # This buffer is never changed in this step. Consequently, no copy is necessary..
-                self._post_transform(in_step)
-                return
+                return self._post_transform(in_step)
 
     @classmethod
     def load(cls, stored_step: dict, inputs, targets, module, file_manager):
diff --git a/pywatts/core/inverse_step.py b/pywatts/core/inverse_step.py
index c29c5af..af4b2a2 100644
--- a/pywatts/core/inverse_step.py
+++ b/pywatts/core/inverse_step.py
@@ -30,4 +30,4 @@ class InverseStep(Step):
             raise KindOfTransformDoesNotExistException(f"The module {self.module.name} has no inverse transform",
                                                        KindOfTransform.INVERSE_TRANSFORM)
 
-        self._post_transform(self.module.inverse_transform(**input_step))
+        return self._post_transform(self.module.inverse_transform(**input_step))
diff --git a/pywatts/core/probabilistic_step.py b/pywatts/core/probabilistic_step.py
index 92fec55..faf4ad6 100644
--- a/pywatts/core/probabilistic_step.py
+++ b/pywatts/core/probabilistic_step.py
@@ -34,4 +34,4 @@ class ProbablisticStep(Step):
             raise KindOfTransformDoesNotExistException(f"The module {self.module.name} has no probablisitic transform",
                                                        KindOfTransform.PROBABILISTIC_TRANSFORM)
 
-        self._post_transform(self.module.predict_proba(input_step))
+        return self._post_transform(self.module.predict_proba(input_step))
diff --git a/pywatts/core/step.py b/pywatts/core/step.py
index 4e43f8f..278c6d0 100644
--- a/pywatts/core/step.py
+++ b/pywatts/core/step.py
@@ -78,8 +78,7 @@ class Step(BaseStep):
             logger.error(message)
             raise NotFittedException(message, self.name, self.module.name)
         result = self.module.transform(**input_step)
-        self._post_transform(result)
-        return result
+        return self._post_transform(result)
 
     @classmethod
     def load(cls, stored_step: Dict, inputs, targets, module, file_manager):
@@ -103,7 +102,7 @@ class Step(BaseStep):
         else:
             train_if = None
         callbacks = []
-        for callback_path in  stored_step["callbacks"]:
+        for callback_path in stored_step["callbacks"]:
             with open(callback_path, 'rb') as pickle_file:
                 callback = cloudpickle.load(pickle_file)
             callback.set_filemanager(file_manager)
@@ -135,7 +134,7 @@ class Step(BaseStep):
         elif self.module is BaseEstimator:
             logger.info("%s not fitted in Step %s", self.module.name, self.name)
 
-        self._transform(input_data)
+        return self._transform(input_data)
 
     def _get_target(self, start, batch):
         return {
diff --git a/pywatts/modules/sample_module.py b/pywatts/modules/sample_module.py
index 481cdf1..f80d6cd 100644
--- a/pywatts/modules/sample_module.py
+++ b/pywatts/modules/sample_module.py
@@ -5,6 +5,7 @@ import xarray as xr
 from pywatts.core.base import BaseTransformer
 from pywatts.core.exceptions.wrong_parameter_exception import WrongParameterException
 from pywatts.utils._xarray_time_series_utils import _get_time_indeces
+import numpy as np
 
 
 class Sampler(BaseTransformer):
@@ -55,7 +56,6 @@ class Sampler(BaseTransformer):
             # Do not use if indexes here, since this would be false if indexes is empty.
             self.indexes = indexes
 
-
     def transform(self, x: xr.DataArray) -> xr.DataArray:
         """
         Sample the given time series x by the lag.
@@ -69,14 +69,11 @@ class Sampler(BaseTransformer):
         if not indexes:
             indexes = _get_time_indeces(x)
         try:
-            result = x
-            for i in range(1, self.sample_size):
-                result = xr.concat([result, x.shift({index: i for index in indexes}, fill_value=0)], dim="horizon")
+            r = [x.shift({index: i for index in indexes}, fill_value=0) for i in range(0, self.sample_size)]
         except ValueError as exc:
             raise WrongParameterException(
                 f"Not all indexes ({indexes}) are in the indexes of x ({list(x.indexes.keys())}).",
                 "Perhaps you set the wrong indexes with set_params or during the initialization of the Sampler.",
                 module=self.name) from exc
-        result = result.transpose(_get_time_indeces(x)[0], "horizon", ...)
-
-        return result
+        result = xr.DataArray(np.stack(r, axis=-1), dims=(*x.dims, "horizon"), coords=x.coords)
+        return result.transpose(_get_time_indeces(x)[0], "horizon", ...)
diff --git a/pywatts/modules/trend_extraction.py b/pywatts/modules/trend_extraction.py
index 181bbeb..f343ca7 100644
--- a/pywatts/modules/trend_extraction.py
+++ b/pywatts/modules/trend_extraction.py
@@ -8,6 +8,7 @@ import xarray as xr
 
 from pywatts.core.base import BaseTransformer
 from pywatts.utils._xarray_time_series_utils import _get_time_indeces
+import numpy as np
 
 
 class TrendExtraction(BaseTransformer):
@@ -79,8 +80,7 @@ class TrendExtraction(BaseTransformer):
         indexes = self.indexes
         if not indexes:
             indexes = _get_time_indeces(x)
-        trend = x.shift({index: self.period for index in indexes}, fill_value=0)
-        for i in range(2, self.length + 1):
-            trend = xr.concat([trend,
-                               x.shift({index: self.period * i for index in indexes}, fill_value=0)], dim="length")
+        trends = [x.shift({index: self.period * i for index in indexes}, fill_value=0) for i in
+                  range(1, self.length + 1)]
+        trend = xr.DataArray(np.stack(trends, axis=-1), dims=(*x.dims, "length"), coords=x.coords)
         return trend.transpose(_get_time_indeces(x)[0], "length", ...)


Test patch:
diff --git a/tests/unit/core/test_inverse_step.py b/tests/unit/core/test_inverse_step.py
index 1872d64..ef4a67f 100644
--- a/tests/unit/core/test_inverse_step.py
+++ b/tests/unit/core/test_inverse_step.py
@@ -36,7 +36,7 @@ class TestInverseTransform(unittest.TestCase):
         self.inverse_module.inverse_transform.assert_called_once_with(input=self.input_step.get_result())
 
     def test_get_result_stop(self):
-        self.input_step._should_stop.return_value = True
+        self.input_step.get_result.return_value = None
         self.inverse_step.get_result(None, None)
 
         self.inverse_module.inverse_transform.assert_not_called()
diff --git a/tests/unit/core/test_probabilistic_step.py b/tests/unit/core/test_probabilistic_step.py
index 3639a28..c4bcb4c 100644
--- a/tests/unit/core/test_probabilistic_step.py
+++ b/tests/unit/core/test_probabilistic_step.py
@@ -29,7 +29,7 @@ class TestProbabilisticStep(unittest.TestCase):
         self.probabilistic_module.predict_proba.assert_called_once_with(input_mock)
 
     def test_get_result_stop(self):
-        self.input_step._should_stop.return_value = True
+        self.input_step.get_result.return_value = None
 
         self.probabilistic_step.get_result(pd.Timestamp("2000.01.01"), pd.Timestamp("2000.01.02"))
 
diff --git a/tests/unit/core/test_start_step.py b/tests/unit/core/test_start_step.py
index 102a521..a44813f 100644
--- a/tests/unit/core/test_start_step.py
+++ b/tests/unit/core/test_start_step.py
@@ -1,4 +1,8 @@
 import unittest
+import pandas as pd
+import xarray as xr
+
+from unittest.mock import MagicMock
 
 from pywatts.core.start_step import StartStep
 
@@ -15,8 +19,28 @@ class TestStartStep(unittest.TestCase):
             "module": "pywatts.core.start_step",
             "class": "StartStep",
             "name": "StartStep",
-            "last":False
+            "last": False
         }
         step = StartStep(None).load(params, None, None, None, None)
         json = step.get_json("file")
         self.assertEqual(params, json)
+
+    def test_get_result_copying(self):
+        # Tests if the get_result method calls correctly the previous step and the module
+
+        input_step = MagicMock()
+        input_step_result_mock = MagicMock()
+        input_step.get_result.return_value = input_step_result_mock
+        input_step._should_stop.return_value = False
+
+        time = pd.date_range('2000-01-01', freq='1H', periods=7)
+        step = StartStep("x")
+        step.buffer = {"x": xr.DataArray([2, 3, 4, 3, 3, 1, 2], dims=["time"],
+                                         coords={'time': time})}
+        result1 = step.get_result(None, None)
+        result2 = step.get_result(None, None)
+
+        # result1 and result2 has to be different objects
+        result1[1] = 20
+
+        self.assertNotEqual(result1[1], result2[1])