Notes

"""
This script will be responsible for parsing the measurement using signal dataframes.

This Dataio module supports following measurement formats till now.

1. *.h5
2. *.mat
3. *.mf4

# import for pip package creation

# from dataio.DataframeConverter import iConverter
# from dataio.SignalDataframe import MeasurementDataframe

"""

import os

import numpy as np
import logging

# from src.dataio.DataframeConverter import iConverter
# from src.dataio.SignalDataframe import MeasurementDataframe
# from dataio.DataframeConverter import iConverter
# from dataio.SignalDataframe import MeasurementDataframe
from DataframeConverter import iConverter
from SignalDataframe import MeasurementDataframe
logger = logging.getLogger(__name__)


class cMeasParser:
def __init__(self, measurement_file_name):
"""
Constructor to find the correct parser for the input measurement file and return it's parser object.
:param measurement_file_name:
"""
self.measurement_file_name = measurement_file_name
file_name, file_extension = os.path.splitext(self.measurement_file_name)
cParser = MeasurementDataframe(self.measurement_file_name, file_extension)
self.parser = cParser
self.dataframe_converter = iConverter()
self.signal_dataframe = None

def get_measurment_dataframe(self):
"""
Get dataframe of input measurement file
:return:
"""
self.signal_dataframe = self.parser.cParser_obj.get_dataframe(self.measurement_file_name)

self.set_dataframe_dtypes()
self.clean_measurement_dataframe()
self.convert_dataframe_to_parquet()
return

def set_dataframe_dtypes(self):
"""
Set dtypes for dataframe
:return:
"""
self.parser.set_dtypes(self.signal_dataframe)
return

def clean_measurement_dataframe(self):
"""
Clean measurement dataframe
:return:
"""
self.parser.clean_dataframe(self.signal_dataframe)
return

def convert_dataframe_to_parquet(self):
"""
Convert generated dataframe into parquet file format
:return:
"""
self.parser.convert_dataframe_to_parquet(self.signal_dataframe)
return

def get_signal_value(self, signal_name):
"""
This function is for reading the signal value from dataframe. This will receive input signal name from
web ui interface.

Signal value reading process for *.mf4 measurement is different than *.mat and *.h5 measurements

:param signal_name -> string type
:return: signal_value -> numpy array
"""
index = self.signal_dataframe.index[self.signal_dataframe.Signal_Name.str.contains(signal_name)][0]

if os.path.splitext(self.measurement_file_name)[1] in ['.MF4', '.mf4']:
signal_value = self.parser.cParser_obj.meas_parser_data.select([signal_name], raw=True)[0].samples
else:
signal_value = np.array(
self.parser.cParser_obj.meas_parser_data[self.signal_dataframe['Signal_Value'].iloc[index]])

return signal_value

def get_signal_time(self, signal_name):
"""
This function is for reading signal time array from dataframe.This will receive input signal name from
web ui interface.

:param signal_name -> string type
:return: signal_time -> numpy array
"""
index = self.signal_dataframe.index[self.signal_dataframe.Signal_Name.str.contains(signal_name)][0]
if os.path.splitext(self.measurement_file_name)[1] in ['.MF4', '.mf4']:
signal_time = self.parser.cParser_obj.meas_parser_data.select([signal_name], raw=True)[0].timestamps
else:
signal_time = np.array(
self.parser.cParser_obj.meas_parser_data[self.signal_dataframe['Signal_Time'].iloc[index]])

return signal_time

def iter_signal_names(self):
"""
This function is for getting the signal names from the dataframe.
:return:
"""
return iter(self.signal_dataframe['Signal_Name'])

def get_signal_names(self):
"""
This function is for getting the signal names from the dataframe.
:return:
"""
signal_names = list(self.signal_dataframe['Signal_Name'])

return signal_names

def get_device_names(self, signal_name):
"""
This function is for getting the device name from the dataframe based on input signal name.

:return:device_name -> string type
"""
index = self.signal_dataframe.index[self.signal_dataframe.Signal_Name.str.contains(signal_name)][0]

device_name = self.signal_dataframe['Device_Name'].iloc[index]

return device_name

def get_measurement_metadata(self, signal_name):
"""
This function is for displaying the measurement metadata information of a selected signal.

:param signal_name -> string type
:return: mesurement metadata info -> dictionary type
"""
pass

def iter_time_keys(self):
"""
Iterate over time axis keys. This is optional i.e. this attribute is not available in all measurement
formats.

:return:
"""
return iter(self.signal_dataframe['NameTimeAxis'])

def iter_device_names(self):
"""
This function is for iterating over device names.

:return:
"""
return iter(self.signal_dataframe['Device_Name'])

def get_signal_length(self, signal_name):
"""
Get length of a selected signal

:param signal_name -> string type
:return: Length of signal -> int or float type
"""
index = self.signal_dataframe.index[self.signal_dataframe.Signal_Name.str.contains(signal_name)][0]

if os.path.splitext(self.measurement_file_name)[1] in ['.MF4', '.mf4']:
signal_length = self.parser.cParser_obj.meas_parser_data.select([signal_name], raw=True)[0].samples.size
else:
signal_length = len(
np.array(self.parser.cParser_obj.meas_parser_data[self.signal_dataframe['Signal_Value'].iloc[index]]))

return signal_length

def is_signal_empty(self, signal_name):
"""
Check whether signal is empty or not

:param signal_name -> string type
:return: True if signal is empty else False
"""
if self.get_signal_length(signal_name) > 0:
return True
else:
return False

def get_physical_unit(self, signal_name):
"""
Physical unit of selected signals

:param signal_name -> string type
:return: signal_unit -> string type
"""
index = self.signal_dataframe.index[self.signal_dataframe.Signal_Name.str.contains(signal_name)][0]

signal_unit = self.signal_dataframe['Unit'].iloc[index]

return signal_unit

def get_time_key(self, signal_name):
"""
Get time key of selected signal

:param signal_name -> string type
:return: Time Key
"""
try:
index = self.signal_dataframe.index[self.signal_dataframe.Signal_Name.str.contains(signal_name)][0]

return self.signal_dataframe['NameTimeAxis'].iloc[index]
except:
logger.info("No time key data available")


if __name__ == '__main__':
# import matplotlib.pyplot as plt

meas_file = r"C:KBData2022-06-30mi5id786__2022-06-30_10-38-08_Resim_23121203.h5"

obj = cMeasParser(meas_file)
obj.get_measurment_dataframe()
signal_name = obj.get_signal_names()[10]
print(obj.get_signal_time(signal_name))
print(obj.get_signal_value(signal_name))
print(obj.get_time_key(signal_name))
print(obj.get_physical_unit(signal_name))
print(obj.get_signal_length(signal_name))
print(obj.get_device_names(signal_name))


# plt.plot(obj.get_signal_time(signal_name), obj.get_signal_value(signal_name), color='C0')
# plt.show()