read data and save it a single variable
This commit is contained in:
Pritimay Sarkar
@@ -5,80 +5,99 @@ from datetime import datetime
|
||||
import math
|
||||
import time
|
||||
from tqdm import tqdm
|
||||
import re
|
||||
|
||||
class Device:
|
||||
def __init__(self, accuracy, precision):
|
||||
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):
|
||||
def perform_hpos_test(repeat_no, solution, df):
|
||||
try:
|
||||
data_file = 'data.xlsx'
|
||||
if os.path.exists(data_file):
|
||||
df = pd.read_excel('data.xlsx')
|
||||
else:
|
||||
df = pd.DataFrame(columns=["sol", "427_buffer_intensity", "555_buffer_intensity", "427_sample_intensity", "555_sample_intensity", "427_absorbance", "555_absorbance", "absorbance_ratio", "time"])
|
||||
|
||||
time.sleep(5)
|
||||
# time.sleep(5)
|
||||
|
||||
port = '/dev/cu.usbserial-23APB136'
|
||||
port = '/dev/cu.usbserial-23APB124'
|
||||
baud_rate = 9600
|
||||
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()
|
||||
print(x)
|
||||
print(s)
|
||||
print(line)
|
||||
|
||||
# perform_blanking = input("Do you want to perform blanking?(Y/N)")
|
||||
|
||||
ser.write(b'I')
|
||||
ser.write(b'B')
|
||||
while True:
|
||||
x = ser.read()
|
||||
b = ser.read(100)
|
||||
line = ser.readline()
|
||||
print(b)
|
||||
if 'SNE' in b.decode("utf-8"):
|
||||
ser.write(b'B')
|
||||
while True:
|
||||
x = ser.read()
|
||||
s = ser.read(100)
|
||||
line = ser.readline()
|
||||
print(s)
|
||||
if 'BC' in s.decode("utf-8"):
|
||||
ser.write(b'P')
|
||||
while True:
|
||||
x = ser.read()
|
||||
r = ser.read(100)
|
||||
line = ser.readline()
|
||||
print(r)
|
||||
if 'RESULT' in r.decode("utf-8") or 'REND' in r.decode('utf-8'):
|
||||
result = r.decode("utf-8").split(' ')
|
||||
df = pd.concat([df, pd.DataFrame([{"sol": solution + '-' + str(repeat_no), "427_buffer_intensity": result[3], "555_buffer_intensity": result[3], "427_sample_intensity": result[5], "555_sample_intensity": result[6], "427_absorbance": math.log10(float(result[3]) / float(result[5])), "555_absorbance": math.log10(float(result[4]) / float(result[6])), "absorbance_ratio": math.log10(float(result[3]) / float(result[5])) / math.log10(float(result[4]) / float(result[6])), "time": datetime.today().strftime('%d-%m-%y %H:%M:%S')}])], ignore_index = True)
|
||||
print(df)
|
||||
writer = pd.ExcelWriter(data_file, engine = 'openpyxl')
|
||||
df.to_excel(writer, sheet_name = 'data', index=False)
|
||||
writer.close()
|
||||
break
|
||||
break
|
||||
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,
|
||||
"time": datetime.today().strftime('%d-%m-%y %H:%M:%S')})
|
||||
break
|
||||
return data_list
|
||||
|
||||
except serial.serialutil.SerialException:
|
||||
print("device is not connected or resuorce busy!")
|
||||
exit()
|
||||
print("Device is not connected or resource busy!")
|
||||
raise
|
||||
except FileNotFoundError as e:
|
||||
print(e)
|
||||
raise
|
||||
except Exception as err:
|
||||
print("error!", err)
|
||||
exit()
|
||||
print("Error:", err)
|
||||
raise
|
||||
finally:
|
||||
if ser is not None:
|
||||
ser.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
data_file = 'data.xlsx'
|
||||
|
||||
d = Device(0.01, 0.001)
|
||||
if os.path.exists(data_file):
|
||||
df = pd.read_excel(data_file)
|
||||
else:
|
||||
df = pd.DataFrame(columns=["sol", "led1Buffer", "led2Buffer", "led3Buffer", "led4Buffer", "time"])
|
||||
|
||||
solution = "tar" #input("solution name: ")
|
||||
repeats = 1 #int(input("number of repeats: "))
|
||||
# device = Device(0.01, 0.001, '/dev/cu.usbserial-23APB124', 9600)
|
||||
|
||||
solution = "buffer" #input("solution name: ")
|
||||
repeats = 100 #int(input("number of repeats: "))
|
||||
for repeat in tqdm(range(0, repeats)):
|
||||
perform_hpos_test(repeat)
|
||||
data_list = perform_hpos_test(repeat, solution, df)
|
||||
df = pd.concat([df, pd.DataFrame(data_list)], ignore_index=True)
|
||||
# print(df)
|
||||
|
||||
|
||||
|
||||
writer = pd.ExcelWriter(data_file, engine = 'openpyxl')
|
||||
df.to_excel(writer, sheet_name = 'data', index=False)
|
||||
writer.close()
|
||||
|
||||
Reference in New Issue
Block a user