300 lines
10 KiB
Python
300 lines
10 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'E')
|
|
while True:
|
|
data += ser.readline().decode("utf-8")
|
|
# print(data.split('\r'))
|
|
if '#RC' in data:
|
|
# print(data)
|
|
# print("DAC values fetched from EPORM")
|
|
break
|
|
|
|
time.sleep(0.5)
|
|
|
|
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'D')
|
|
while True:
|
|
data += ser.readline().decode("utf-8")
|
|
# print(data.split('\r'))
|
|
if '#DC' in data:
|
|
print(data)
|
|
print("D_CMD process completed")
|
|
break
|
|
|
|
time.sleep(0.5)
|
|
|
|
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
|
|
|
|
time.sleep(0.5)
|
|
|
|
ser.write(b'G')
|
|
while True:
|
|
data += ser.readline().decode("utf-8")
|
|
# print(data.split('\r'))
|
|
if '#WC' in data:
|
|
print(data)
|
|
print("DAC values saved to EPROM")
|
|
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 diagnostics_and_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_autodac = pd.read_excel(data_file, sheet_name=sheet_name)
|
|
else:
|
|
print(f"Sheet '{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"])
|
|
|
|
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_autodac)
|
|
|
|
df_autodac = pd.concat([df_autodac, pd.DataFrame(data_list)], ignore_index=True)
|
|
|
|
writer = pd.ExcelWriter(data_file, engine = 'openpyxl')
|
|
df_autodac.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()
|
|
|
|
diagnostics_and_autodac_process()
|
|
|
|
buffer_process() |