chore: логика поиска интервалов перенесена на более эффективную библиотеку intervaltree

src/performance/performance.py

@@ -2,6 +2,7 @@ from typing import Optional
 import os
 import numpy as np
 import pandas as pd
+from intervaltree import Interval, IntervalTree
 
 from roboter import RobotPerformance, TWC_Performance
 
@@ -10,8 +11,25 @@ class PerformanceProcessor:
     def calc_performance(self, path, df):
         robot = RobotPerformance()
         TWC = TWC_Performance()
-        rob_df = robot.job(path)
-        TWC_df = TWC.job(df)
+        point_tree, movement_tree, dialog_tree = robot.job(path)
+        closing_tree, squeeze_tree, relief_tree = TWC.job(df)
+
+        dialog_inPoint = point_tree & dialog_tree
+        dialog_inMovement = movement_tree & dialog_tree
+
+        closing_inPoint = point_tree & closing_tree
+        closing_inMovement = movement_tree & closing_tree
+
+        squeeze_inPoint = point_tree & squeeze_tree
+        squeeze_inMovement = movement_tree & squeeze_tree
+        
+        relief_inPoint = point_tree & relief_tree
+        relief_inMovement = movement_tree & relief_tree
+
+
+
+        
+
 
 
 
@@ -20,4 +38,5 @@ if __name__ == "__main__":
     robot = RobotPerformance()
     #TWC = TWC_Performance()
     result = robot.job(path)
-    print (result)
+    print (result[0])
+    print (result[1])

src/performance/roboter.py

@@ -5,6 +5,8 @@ import os
 import numpy as np
 import pandas as pd
 
+from intervaltree import Interval, IntervalTree
+
 
 class BasePerformanceFactory(ABC):
 
@@ -88,16 +90,35 @@ class BaseProduct(ABC):
         stage_diff = np.diff(dataframe[signal])
         start_idx = np.where(stage_diff == 1)
         finish_idx = np.where(stage_diff == -1)
-        return start_idx, finish_idx
+        return start_idx[0], finish_idx[0]
     
-    def operation(self, dataframe: pd.DataFrame) -> pd.DataFrame: 
-        all_idx = np.array([])
-        for signal in self._signals:
+    def _find_events(self, 
+                        dataframe: pd.DataFrame, 
+                        signals: list[str]) -> Optional[dict[dict[pd.Series]]]: 
+        intervals = {}
+        end_time = 0
+        for signal in signals:
             start_idx, finish_idx = np.array(self._find_indexes(signal, dataframe))
-            all_idx = np.hstack([all_idx, start_idx[0], finish_idx[0]])
-        all_idx = np.sort(np.array(all_idx, dtype="int64"))
-        result = dataframe.loc[all_idx, self._signals + ["time"]]
-        return result
+            start_series = dataframe.loc[start_idx, "time"].reset_index(drop=True)
+            end_series = dataframe.loc[finish_idx, "time"].reset_index(drop=True)
+            end_series.fillna(end_time)
+            intervals[signal] = {"rise": start_series,
+                                "fall": end_series}
+        return intervals
+    
+    def _form_intervals(self, 
+                        start: pd.Series, 
+                        end: pd.Series) -> IntervalTree:
+        if len(start) != len(end):
+            for i in range(1, len(end)):
+                if end[i-1] > start[i]:
+                   start = start.drop(i).reset_index(drop=True)
+        tree = IntervalTree(Interval(start[i], end[i], i) for i in range(len (start)))
+        return tree
+    
+    @abstractmethod
+    def operation(self):
+        ...
 
 
 class RobotPerformance(BasePerformanceFactory):
@@ -105,33 +126,18 @@ class RobotPerformance(BasePerformanceFactory):
     def factory_method(self) -> BaseProduct:
         return RobotDF()
     
-    def job(self, path) -> pd.DataFrame:
+    def job(self, path) -> list[pd.DataFrame]:
         product = self.factory_method()
         dataframe = self._get_file_data(path)
         rob_df = product.operation(dataframe)
-        in_point, in_moving, in_dialog = False, False, False
-        for index, row in rob_df.iterrows():
-            if row["$OUT3012"] == 1:
-                if not in_point:
-                    in_point = True
-                    start_point = row["time"]
-                else:
-                    in_point = False
-                    time_in_point = row["time"] - start_point
-            
-
-
-
         return rob_df
-    
-
 
 class TWC_Performance(BasePerformanceFactory):
 
     def factory_method(self) -> BaseProduct:
         return TWC_DF()
     
-    def job(self, TWC_DF: pd.DataFrame) -> pd.DataFrame:
+    def job(self, TWC_DF: pd.DataFrame) -> list[pd.DataFrame]:
         product = self.factory_method()
         result = product.operation(TWC_DF) 
         return result
@@ -145,6 +151,13 @@ class RobotDF(BaseProduct):
             "$OUT3003"
         ]
 
+    def operation(self, dataframe: pd.DataFrame) -> list[IntervalTree]:
+        events = self._find_events(dataframe, self._signals)
+        point_tree = self._form_intervals(start=events["$OUT3012"]["rise"], end=events["$OUT3012"]["fall"])
+        movement_tree = self._form_intervals(start=events["$OUT3003"]["fall"], end=events["$OUT3012"]["rise"])
+        dialog_tree = []
+        return point_tree, movement_tree, dialog_tree
+
     
 class TWC_DF(BaseProduct):
     def __init__(self):
@@ -155,6 +168,13 @@ class TWC_DF(BaseProduct):
             "Relief"
         ]
 
+    def operation(self, dataframe: pd.DataFrame) -> list[IntervalTree]:
+        events = self._find_events(dataframe, self._signals)
+        closing_tree = self._form_intervals(start=events["Closing"]["rise"], end=events["Closing"]["fall"])
+        squeeze_tree = self._form_intervals(start=events["Squeeze"]["rise"], end=events["Squeeze"]["fall"])
+        relief_tree = self._form_intervals(start=events["Relief"]["rise"], end=events["Relief"]["fall"])
+        return closing_tree, squeeze_tree, relief_tree
+