WeldingSpotPerformance/src/utils/diagram_parser.py

import pandas as pd
import numpy as np

class DiagramParser:
    def __init__(self, system_config: dict):
        self.data = pd.DataFrame({})
        self.signals = [system_config["Closure_signal"],
                        system_config["Squeeze_signal"],
                        system_config["Welding_signal"],
                        system_config["Release_signal"]]
        
        self.boolDict = {}
        self.floatDict = {}
        self.timingsDict = {}
        self.theor_mode = False

    def setData(self, path):
        if not path:
            self.theor_mode = True
        else:
            self.data = pd.read_csv(path)
            
            for signalName in self.data.columns:
                if type (self.data[signalName].iloc[0]) == np.bool:
                    self.boolDict[signalName] = self._getBoolChanges(signalName)

            for signalName in self.data.columns:
                if type (self.data[signalName].iloc[0]) == np.float64:
                    self.floatDict[signalName] = self._getFloatChanges(signalName)
            
            for key, items in self.boolDict.items():
                if key == self.signals[0]: name = "closure"
                elif key == self.signals[1]: name =  "compression"
                elif key == self.signals[2]: name =  "welding"
                elif key == self.signals[3]: name =  "opening"

                self.timingsDict[name] = []
                for i, item in enumerate(items):
                    if item[1] == 'high':
                        self.timingsDict[name].append([items[i][0], items[i+1][0]])
        
    def getBoolDict (self) -> dict:
        return self.boolDict
    
    def getFloatDict (self) -> dict:
        return self.floatDict
    
    def getRealTimings(self) -> dict:
        return self.timingsDict
    
    def getMode(self) -> bool:
        return self.theor_mode
    
    def _getBoolChanges(self, signalName) -> list:
        timeCode = self.data['time']
        signal_values = self.data[signalName]
        changes = []

        if len(signal_values) > 0:
            changes.append([float(timeCode.iloc[0]), 'high' if bool(signal_values.iloc[0]) else 'low'])

            for i in range(1, len(signal_values)):
                if signal_values.iloc[i] != signal_values.iloc[i - 1]:
                    changes.append([float(timeCode.iloc[i]), 'high' if bool(signal_values.iloc[i]) else 'low'])
        
        return changes
    

    def _getFloatChanges(self, signalName) -> list:
        timeCode = self.data['time']
        signal_values = self.data[signalName]
        changes = []

        if len(signal_values) > 0:
            for i in range (len(signal_values)):
                changes.append([float(timeCode.iloc[i]), float(signal_values.iloc[i])])
        
        return changes