add prototype to scale with anomalies

Author: arch

funscript_editor/algorithms/funscriptgenerator.py

@@ -240,6 +240,10 @@ def calculate_score(self) -> None:
             self.score_x = [max([x[0] for x in self.bboxes['Woman']]) - w[0] for w in self.bboxes['Woman']]
             self.score_y = [max([x[1] for x in self.bboxes['Woman']]) - w[1] for w in self.bboxes['Woman']]
 
+        # NOTE: Scaling with the anomaly function sometimes makes the result even worse
+        # self.score_x = sp.scale_signal_with_anomalies(self.score_x, 0, 100, lower_quantile=0.05, upper_quantile=0.99999)
+        # self.score_y = sp.scale_signal_with_anomalies(self.score_y, 0, 100, lower_quantile=0.05, upper_quantile=0.99999)
+
         self.score_x = sp.scale_signal(self.score_x, 0, 100)
         self.score_y = sp.scale_signal(self.score_y, 0, 100)
 
@@ -504,6 +508,7 @@ def tracking(self) -> str:
 
         video.stop()
         self.__logger.info(status)
+        self.calculate_score()
         return status
 
 
@@ -562,7 +567,6 @@ def run(self) -> None:
         # NOTE: score_y and score_x should have the same number of elements so it should be enouth to check one score length
         with Listener(on_press=self.on_key_press) as listener:
             status = self.tracking()
-            self.calculate_score()
             if len(self.score_y) >= HYPERPARAMETER['min_frames']:
                 if self.params.direction != 'x':
                     self.scale_score(status, direction='y')

funscript_editor/algorithms/signalprocessing.py

@@ -17,6 +17,33 @@ def scale_signal(signal :list, lower: float = 0, upper: float = 99) -> list:
     return [(upper - lower) * (x - min(signal)) / (max(signal) - min(signal)) + lower for x in signal]
 
 
+def scale_signal_with_anomalies(
+        signal :list,
+        lower: float = 0,
+        upper: float = 99,
+        lower_quantile: float = 0.0005,
+        upper_quantile: float = 0.9995) -> list:
+    """ Scale an signal (list of float or int) between given lower and upper value
+
+    Args:
+        signal (list): list with float or int signal values to scale
+        lower (float): lower scale value
+        upper (float): upper scale value
+        lower_quantile (float): lower quantile value to filter [0,1]
+        upper_quantile (float): upper quantile value to filter [0,1]
+
+    Returns:
+        list: list with scaled signal
+    """
+    a1 = np.quantile(signal, lower_quantile)
+    a2 = np.quantile(signal, upper_quantile)
+    anomaly_free = np.array([x for x in signal if a1 < x < a2])
+    anomaly_free_min = min(anomaly_free)
+    anomaly_free_max = max(anomaly_free)
+    scaled = [(upper - lower) * (x - anomaly_free_min) / (anomaly_free_max - anomaly_free_min) + lower for x in signal]
+    return [min((anomaly_free_max, max((anomaly_free_min, x)) )) for x in scaled]
+
+
 def moving_average(x :list, w: int) -> list:
     """ Calculate moving average for given signal x with window size w