WeldingSpotPerformance/src/performance/performance.py

from typing import Optional, Tuple

import numpy as np
import pandas as pd

   
#TODO: Реализовать поиск времени простоя
class DowntimeAnalyzer:
    def __init__(self) -> None:
        ...

    def generate_downtime_report(self) -> list:
        ...

    def _separate_conditions(self, TWC_raw:pd.DataFrame, robot_raw:pd.DataFrame) -> Tuple[dict, dict]:
        robot_splitter = RobotConditionSplitter()
        TWC_splitter = TWC_ConditionSplitter()
        #comm_splitter = CommConditionSplitter()
        rob_conditions = robot_splitter.split(robot_raw)
        TWC_conditions = TWC_splitter.split(TWC_raw)
        #comm_df = comm_splitter.split(rob_conditions["communication"], TWC_conditions["communication"])
        return (rob_conditions, TWC_conditions)


class ConditionSplitter:
    def __init__(self):
        self._signals = []

    def _find_indexes(self,
                       signal: str,  
                       dataframe: pd.DataFrame) -> list[list[float], list[float]]:
        stage_diff = np.diff(dataframe[signal])
        start_idx = np.where(stage_diff == 1)
        finish_idx = np.where(stage_diff == -1)
        return start_idx[0], finish_idx[0]
    
    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))
            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) -> dict:
        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)
        intervals = {'start':start.tolist, 'end':end.tolist}
        return intervals
    

class RobotConditionSplitter(ConditionSplitter):
    def __init__(self):
        self._signals = [
            "$OUT3012", 
            "$IN3003", 
            "$OUT3003", 
            "$OUT3244"
        ]

    def split(self, dataframe: pd.DataFrame) -> dict:
        events = self._find_events(dataframe, self._signals)
        communication_sig = {'sent':events["$OUT3244"]["fall"], 'received':events[""][""]}
        point_interval = self._form_intervals(start=events["$OUT3012"]["rise"], end=events["$OUT3012"]["fall"])
        movement_interval = self._form_intervals(start=events["$OUT3244"]["rise"], end=events["$OUT3244"]["fall"])
        conditions = {"communication":communication_sig, "waiting": point_interval, "moving": movement_interval}
        return conditions

    
class TWC_ConditionSplitter(ConditionSplitter):
    def __init__(self):
        self._signals = [
            "Closing",
            "Squeeze",
            "Welding",
            "Relief", 
            "Oncoming"
        ]

    def split(self, dataframe: pd.DataFrame) -> dict:
        events = self._find_events(dataframe, self._signals)
        communication_sig = {'sent':events[""][""], 'received':events[""][""]} #ситара что-то делает -конец сигнала
        closing_interval = self._form_intervals(start=events["Closing"]["rise"], end=events["Closing"]["fall"])
        squeeze_interval = self._form_intervals(start=events["Squeeze"]["rise"], end=events["Squeeze"]["fall"])
        relief_interval = self._form_intervals(start=events["Relief"]["rise"], end=events["Relief"]["fall"])
        oncoming_interval = self._form_intervals(start=events["Oncoming"]["rise"], end=events["Oncoming"]["fall"])
        conditions = {
            "communication":communication_sig, 
            'closing':closing_interval,
            'squeeze':squeeze_interval,
            'relief':relief_interval,
            'oncoming':oncoming_interval
            }
        return conditions


class CommConditionSplitter(ConditionSplitter):
    """
    Определяет промежуток, в который происходит взаимодействие между нодами.
    """

    def split(self, node1: dict, node2: dict) -> pd.DataFrame:
        n1_to_n2 = self._form_intervals(start=node1['sent'], end=node2['received'])
        n2_to_n1 = self._form_intervals(start=node2['sent'], end=node1['received'])
        return pd.concat([pd.DataFrame(n1_to_n2), pd.DataFrame(n2_to_n1)])
    

if __name__ == "__main__":
    path = "D:\downloads\Test2\TeslaTIME29_71_KRCIO.dat"