Files
hpos-data/scripts/quality_control_leds.py
Pritimay Sarkar 95d628f38d add D command
2024-03-05 16:02:41 +05:30

266 lines
9.1 KiB
Python

import serial
import pandas as pd
import os.path
from datetime import datetime
import math
from tqdm import tqdm
import re
import sys
import time
class Device:
def __init__(self, accuracy, precision, port, baud_rate):
self.accuracy = accuracy
self.precision = precision
self.port = port
self.baud_rate = 9600
def perform_hpos_test(repeat_no, solution, df):
try:
port = sys.argv[1]
baud_rate = 9600
deviceId = 'QC'
data = ""
data_list = []
if not os.path.exists(port):
raise FileNotFoundError(f"Port {port} not found.")
with serial.Serial(port, baud_rate, timeout=1, parity=serial.PARITY_NONE) as ser:
x = ser.read()
s = ser.read(10)
line = ser.readline()
ser.write(b'B')
while True:
data += ser.readline().decode("utf-8")
# print(data.split('\r'))
# if 'SNE' in data:
# ser.write(b'B')
if '#BC' in data:
ser.write(b'P')
if 'REND' in data:
# print(data)
pattern = re.compile(r'(LB[1-4])\s+([\d.]+)')
matches = pattern.findall(data)
if matches:
lb1_value = matches[0][1]
lb2_value = matches[1][1]
lb3_value = matches[2][1]
lb4_value = matches[3][1]
# print("LB1:", lb1_value)
# print("LB2:", lb2_value)
# print("LB3:", lb3_value)
# print("LB4:", lb4_value)
else:
print("Pattern not found.")
data_list.append({"sol": solution + '-' + str(repeat_no),
"led1Buffer": lb1_value,
"led2Buffer": lb2_value,
"led3Buffer": lb3_value,
"led4Buffer": lb4_value,
"deviceId": deviceId,
"time": datetime.today().strftime('%d-%m-%y %H:%M:%S')})
break
return data_list
except serial.serialutil.SerialException:
print("Device is not connected or resource busy!")
raise
except FileNotFoundError as e:
print(e)
raise
except Exception as err:
print("Error:", err)
raise
finally:
if ser is not None:
# print("con close")
ser.close()
def perform_auto_dac(df):
try:
port = sys.argv[1]
baud_rate = 9600
deviceId = 'QC'
data = ""
data_list = []
if not os.path.exists(port):
raise FileNotFoundError(f"Port {port} not found.")
with serial.Serial(port, baud_rate, timeout=1, parity=serial.PARITY_NONE) as ser:
x = ser.read()
s = ser.read(10)
line = ser.readline()
ser.write(b'C')
while True:
data += ser.readline().decode("utf-8")
# print(data.split('\r'))
if '#CC' in data:
# print(data)
pattern = re.compile(r'LED(\d+)_DAC: (\d+)')
matches = pattern.findall(data)
if matches:
lb1_value = matches[0][1]
lb2_value = matches[1][1]
lb3_value = matches[2][1]
lb4_value = matches[3][1]
print("LED1_DAC:", lb1_value)
print("LED2_DAC:", lb2_value)
print("LED3_DAC:", lb3_value)
print("LED4_DAC:", lb4_value)
else:
print("Pattern not found.")
data_list.append({"LED1_DAC": lb1_value,
"LED2_DAC": lb2_value,
"LED3_DAC": lb3_value,
"LED4_DAC": lb4_value,
"deviceId": deviceId,
"time": datetime.today().strftime('%d-%m-%y %H:%M:%S')})
break
return data_list
except serial.serialutil.SerialException:
print("Device is not connected or resource busy!")
raise
except FileNotFoundError as e:
print(e)
raise
except Exception as err:
print("Error:", err)
raise
finally:
if ser is not None:
# print("con close")
ser.close()
def autodac_process():
data_file = 'quality_control_leds.xlsx'
sheet_name = 'auto_dac'
if os.path.exists(data_file):
xls = pd.ExcelFile(data_file)
if sheet_name in xls.sheet_names:
df = pd.read_excel(data_file, sheet_name=sheet_name)
else:
print(f"Sheet '{sheet_name}' does not exist in the Excel file.")
df = pd.DataFrame(columns=["LED1_DAC", "LED2_DAC", "LED3_DAC", "LED4_DAC", "deviceId", "time"])
else:
df = pd.DataFrame(columns=["LED1_DAC", "LED2_DAC", "LED3_DAC", "LED4_DAC", "deviceId", "time"])
buffer_sheet_name = 'buffer'
if os.path.exists(data_file):
xls = pd.ExcelFile(data_file)
if buffer_sheet_name in xls.sheet_names:
df_buffer = pd.read_excel(data_file, sheet_name=buffer_sheet_name)
else:
print(f"Sheet '{buffer_sheet_name}' does not exist in the Excel file.")
df_buffer = pd.DataFrame(columns=["LED1_DAC", "LED2_DAC", "LED3_DAC", "LED4_DAC", "deviceId", "time"])
else:
df_buffer = pd.DataFrame(columns=["sol", "led1Buffer", "led2Buffer", "led3Buffer", "led4Buffer", "deviceId", "time"])
data_list = perform_auto_dac(df)
df = pd.concat([df, pd.DataFrame(data_list)], ignore_index=True)
writer = pd.ExcelWriter(data_file, engine = 'openpyxl')
df.to_excel(writer, sheet_name = sheet_name, index=False)
df_buffer.to_excel(writer, sheet_name = buffer_sheet_name, index=False)
writer.close()
time.sleep(1)
def buffer_process():
data_file = 'quality_control_leds.xlsx'
sheet_name = 'buffer'
autodac_sheet_name = 'auto_dac'
if os.path.exists(data_file):
xls = pd.ExcelFile(data_file)
if sheet_name in xls.sheet_names:
df = pd.read_excel(data_file, sheet_name=sheet_name)
else:
print(f"Sheet '{sheet_name}' does not exist in the Excel file.")
df = pd.DataFrame(columns=["sol", "led1Buffer", "led2Buffer", "led3Buffer", "led4Buffer", "deviceId", "time"])
else:
df = pd.DataFrame(columns=["sol", "led1Buffer", "led2Buffer", "led3Buffer", "led4Buffer", "deviceId", "time"])
if os.path.exists(data_file):
xls = pd.ExcelFile(data_file)
if autodac_sheet_name in xls.sheet_names:
df_autodac = pd.read_excel(data_file, sheet_name=autodac_sheet_name)
else:
print(f"Sheet '{autodac_sheet_name}' does not exist in the Excel file.")
df_autodac = pd.DataFrame(columns=["LED1_DAC", "LED2_DAC", "LED3_DAC", "LED4_DAC", "deviceId", "time"])
else:
df_autodac = pd.DataFrame(columns=["LED1_DAC", "LED2_DAC", "LED3_DAC", "LED4_DAC", "deviceId", "time"])
print(df_autodac)
solution = "buffer" #input("solution name: ")
repeats = 5 #int(input("number of repeats: "))
for repeat in tqdm(range(0, repeats)):
data_list = perform_hpos_test(repeat, solution, df)
df = pd.concat([df, pd.DataFrame(data_list)], ignore_index=True)
writer = pd.ExcelWriter(data_file, engine = 'openpyxl')
df.to_excel(writer, sheet_name = sheet_name, index=False)
df_autodac.to_excel(writer, sheet_name = autodac_sheet_name, index=False)
writer.close()
def diagnostics_process():
try:
port = sys.argv[1]
baud_rate = 9600
deviceId = 'QC'
data = ""
data_list = []
if not os.path.exists(port):
raise FileNotFoundError(f"Port {port} not found.")
with serial.Serial(port, baud_rate, timeout=1, parity=serial.PARITY_NONE) as ser:
x = ser.read()
s = ser.read(10)
line = ser.readline()
ser.write(b'D')
while True:
data += ser.readline().decode("utf-8")
# print(data.split('\r'))
if '#DC' in data:
print(data)
print("D_CMD process completed")
return
return
except serial.serialutil.SerialException:
print("Device is not connected or resource busy!")
raise
except FileNotFoundError as e:
print(e)
raise
except Exception as err:
print("Error:", err)
raise
finally:
if ser is not None:
ser.close()
if __name__ == "__main__":
# device = Device(0.01, 0.001, '/dev/cu.usbserial-23APB124', 9600)
diagnostics_process()
autodac_process()
buffer_process()