Recalculate input if a module is refitted.

pywatts/core/base_step.py

@@ -48,7 +48,7 @@ def __init__(self, input_steps: Optional[Dict[str, "BaseStep"]] = None,
         self.transform_time = SummaryObjectList(self.name + " Transform Time", category=SummaryCategory.TransformTime)
 
     def get_result(self, start: pd.Timestamp, end: Optional[pd.Timestamp], buffer_element: str = None,
-                   return_all=False, minimum_data=(0, pd.Timedelta(0))):
+                   return_all=False, minimum_data=(0, pd.Timedelta(0)), recalculate=False):
         """
         This method is responsible for providing the result of this step.
         Therefore,
@@ -71,9 +71,10 @@ def get_result(self, start: pd.Timestamp, end: Optional[pd.Timestamp], buffer_el
             return None
 
         # 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 recalculate or (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, minimum_data)
+                self._compute(start, end, minimum_data, recalculate)
                 self._current_end = end
             if not end:
                 self.finished = True
@@ -86,7 +87,7 @@ def get_result(self, start: pd.Timestamp, end: Optional[pd.Timestamp], buffer_el
 
         return self._pack_data(start, end, buffer_element, return_all=return_all, minimum_data=minimum_data)
 
-    def _compute(self, start, end, minimum_data) -> Dict[str, xr.DataArray]:
+    def _compute(self, start, end, minimum_data, recalculate=False) -> Dict[str, xr.DataArray]:
         pass
 
     def further_elements(self, counter: pd.Timestamp) -> bool:
@@ -197,10 +198,10 @@ def load(cls, stored_step: dict, inputs, targets, module, file_manager):
         :return: The restored step.
         """
 
-    def _get_input(self, start, batch, minimum_data=(0, pd.Timedelta(0))):
+    def _get_input(self, start, batch, minimum_data=(0, pd.Timedelta(0)), recalculate=False):
         return None
 
-    def _get_target(self, start, batch, minimum_data=(0, pd.Timedelta(0))):
+    def _get_target(self, start, batch, minimum_data=(0, pd.Timedelta(0)), recalculate=False):
         return None
 
     def _should_stop(self, start, end) -> bool:

pywatts/core/either_or_step.py

@@ -15,22 +15,21 @@ def __init__(self, input_steps):
         super().__init__(input_steps)
         self.name = "EitherOr"
 
-    def _compute(self, start, end, minimum_data):
-        input_data = self._get_input(start, end, minimum_data)
+    def _compute(self, start, end, minimum_data, recalculate=False):
+        input_data = self._get_input(start, end, minimum_data, recalculate=recalculate)
         return self._transform(input_data)
 
-    def _get_input(self, start, batch, minimum_data=(0, pd.Timedelta(0))):
+    def _get_input(self, start, batch, minimum_data=(0, pd.Timedelta(0)), recalculate=False):
         inputs = []
         for step in self.input_steps.values():
-            inp = step.get_result(start, batch, minimum_data=minimum_data)
+            inp = step.get_result(start, batch, minimum_data=minimum_data, recalculate=recalculate)
             inputs.append(inp)
         return inputs
 
     def _transform(self, input_step):
         # Chooses the first input_step which calculation is not stopped.
         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..
                 return self._post_transform(in_step)
 
     @classmethod

pywatts/core/inverse_step.py

@@ -29,5 +29,4 @@ def _transform(self, input_step):
         if not self.module.has_inverse_transform:
             raise KindOfTransformDoesNotExistException(f"The module {self.module.name} has no inverse transform",
                                                        KindOfTransform.INVERSE_TRANSFORM)
-
         return self._post_transform(self.module.inverse_transform(**input_step))

pywatts/core/probabilistic_step.py

@@ -33,5 +33,4 @@ def _transform(self, input_step):
         if not self.module.has_predict_proba:
             raise KindOfTransformDoesNotExistException(f"The module {self.module.name} has no probablisitic transform",
                                                        KindOfTransform.PROBABILISTIC_TRANSFORM)
-
         return self._post_transform(self.module.predict_proba(input_step))

pywatts/core/result_step.py

@@ -16,7 +16,7 @@ def __init__(self, input_steps, buffer_element: str):
         self.buffer_element = buffer_element
 
     def get_result(self, start: pd.Timestamp, end: Optional[pd.Timestamp], buffer_element: str = None,
-                   return_all=False, minimum_data=(0, pd.Timedelta(0))):
+                   return_all=False, minimum_data=(0, pd.Timedelta(0)), recalculate=False):
         """
         Returns the specified result of the previous step.
         """

pywatts/core/step.py

@@ -145,22 +145,14 @@ def load(cls, stored_step: Dict, inputs, targets, module, file_manager):
 
         return step
 
-    def _compute(self, start, end, minimum_data):
-        input_data = self._get_input(start, end, minimum_data)
-        target = self._get_target(start, end, minimum_data)
+    def _compute(self, start, end, minimum_data, recalculate=False):
+        input_data = self._get_input(start, end, minimum_data, recalculate=recalculate)
+        target = self._get_target(start, end, minimum_data, recalculate=recalculate)
         if self.current_run_setting.computation_mode in [ComputationMode.Default, ComputationMode.FitTransform,
                                                          ComputationMode.Train]:
-            # Fetch input_data and target data
-            if self.batch_size:
-                input_batch = self._get_input(end - self.batch_size, end, minimum_data)
-                target_batch = self._get_target(end - self.batch_size, end, minimum_data)
-                start_time = time.time()
-                self._fit(input_batch, target_batch)
-                self.training_time.set_kv("", time.time() - start_time)
-            else:
-                start_time = time.time()
-                self._fit(input_data, target)
-                self.training_time.set_kv("", time.time() - start_time)
+            start_time = time.time()
+            self._fit(input_data, target)
+            self.training_time.set_kv("", time.time() - start_time)
         elif self.module is BaseEstimator:
             logger.info("%s not fitted in Step %s", self.module.name, self.name)
 
@@ -174,7 +166,7 @@ def _compute(self, start, end, minimum_data):
             result_dict[key] = res.sel(**{_get_time_indexes(res)[0]: index[(index >= start)]})
         return result_dict
 
-    def _get_target(self, start, batch, minimum_data=(0, pd.Timedelta(0))):
+    def _get_target(self, start, batch, minimum_data=(0, pd.Timedelta(0)), recalculate=False):
         min_data_module = self.module.get_min_data()
         if isinstance(min_data_module, (int, np.integer)):
             minimum_data = minimum_data[0] + min_data_module, minimum_data[1]
@@ -185,14 +177,14 @@ def _get_target(self, start, batch, minimum_data=(0, pd.Timedelta(0))):
             for key, target in self.targets.items()
         }
 
-    def _get_input(self, start, batch, minimum_data=(0, pd.Timedelta(0))):
+    def _get_input(self, start, batch, minimum_data=(0, pd.Timedelta(0)), recalculate=False):
         min_data_module = self.module.get_min_data()
         if isinstance(min_data_module, (int, np.integer)):
             minimum_data = minimum_data[0] + min_data_module, minimum_data[1]
         else:
             minimum_data = minimum_data[0], minimum_data[1] + min_data_module
         return {
-            key: input_step.get_result(start, batch, minimum_data=minimum_data) for
+            key: input_step.get_result(start, batch, minimum_data=minimum_data, recalculate=False) for
             key, input_step in self.input_steps.items()
         }
 
@@ -237,15 +229,15 @@ def refit(self, start: pd.Timestamp, end: pd.Timestamp):
                         self._refit(end)
                         break
                 elif isinstance(refit_condition, Callable):
-                    input_data = self._get_input(start, end)
-                    target = self._get_target(start, end)
+                    input_data = self._get_input(start, end, recalculate=False)
+                    target = self._get_target(start, end, recalculate=False)
                     if refit_condition(input_data, target):
                         self._refit(end)
                         break
 
     def _refit(self, end):
-        refit_input = self._get_input(end - self.retrain_batch, end)
-        refit_target = self._get_target(end - self.retrain_batch, end)
+        refit_input = self._get_input(end - self.retrain_batch, end, recalculate=True)
+        refit_target = self._get_target(end - self.retrain_batch, end, recalculate=True)
         self.module.refit(**refit_input, **refit_target)
 
     def get_result_step(self, item: str):