132 lines
4.4 KiB
Python
132 lines
4.4 KiB
Python
from ui import Frame
|
|
from datetime import datetime
|
|
from threading import *
|
|
import os, json, sys, wx, subprocess
|
|
import serial.tools.list_ports
|
|
import numpy as np
|
|
|
|
HEIGHT = 800
|
|
WIDTH = 800
|
|
|
|
|
|
def main():
|
|
#################### USER INPUTS ###########################
|
|
# TODO: if there is already a setting.json exist, loads all inputs from there as 'default'
|
|
# 9840
|
|
resistors = [float(frame.r_ref_1.GetValue()), float(frame.r_ref_2.GetValue()), float(frame.r_ref_3.GetValue()),
|
|
float(frame.r_ref_4.GetValue()),
|
|
0] # resisters for each An port, where n is an integer from 0-3. Use 0 if none. in Ohms
|
|
input_voltage = float(frame.input_voltage.GetValue())
|
|
adjusted_volt = float(frame.adjusted_volt.GetValue())
|
|
port = frame.dev_list.GetValue()
|
|
# typical window size: 480
|
|
window_size = int(frame.m_textCtrl26.GetValue())
|
|
|
|
#################### END USER INPUTS ########################
|
|
|
|
# filename, disable customize name but should inform the user the current name of the data file on the front end
|
|
# alternatively, could be default input if no other input given. But if there is might have problem...
|
|
dat_folder = "RecordedData"
|
|
os.makedirs(dat_folder, exist_ok=True)
|
|
filename = os.path.join(os.getcwd(), dat_folder, f"{datetime.now().strftime('%Y-%m-%d_%H%M%S')}.csv")
|
|
resolution = 1023 # Arduino's analogRead()'s input resolution. We don't change this number usually.
|
|
delay = 1000 # millisec pe r data, defined in the firmware
|
|
if not (len(resistors) == 5):
|
|
raise ValueError(f"expecting 5 resistor values, but got {len(resistors)}!!!")
|
|
|
|
gen_settings(resistors, input_voltage, adjusted_volt, port, filename, resolution, window_size, delay)
|
|
return filename
|
|
|
|
|
|
def gen_settings(resistors, input_voltage, adjusted_volt, port, filename, resolution, window_size, delay):
|
|
"""
|
|
export all inputs from main() to a .json file
|
|
"""
|
|
name = "settings.json"
|
|
settings = {}
|
|
settings["refRes"] = resistors
|
|
settings["sensor_ports"] = np.where(np.array(resistors) > 0)[0].tolist()
|
|
settings["v_src"] = input_voltage
|
|
settings["v_in"] = adjusted_volt
|
|
settings["port"] = port
|
|
settings["resolution"] = resolution
|
|
settings["winSize"] = window_size
|
|
settings["file_name"] = filename
|
|
settings["delay"] = delay
|
|
open(name, 'w').write(json.dumps(settings, indent=4))
|
|
open(filename, "a", newline="", encoding="utf-8")
|
|
|
|
|
|
def get_devices():
|
|
"""
|
|
get all available devices connected, put them in the UI's dropdown menu
|
|
"""
|
|
ports = [comport.device for comport in serial.tools.list_ports.comports()] # get all available ports
|
|
frame.dev_list.AppendItems(ports)
|
|
|
|
|
|
# return ports
|
|
|
|
|
|
# for scheduler maybe?
|
|
def task1():
|
|
print("run task 1")
|
|
run_t1 = ["start", sys.executable, "read_arduino.pyw"]
|
|
out1 = subprocess.Popen(run_t1, shell=True, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
running = True
|
|
init = 0
|
|
while running:
|
|
cline = out1.stdout.readline()
|
|
print(cline.decode())
|
|
if not cline and init > 0:
|
|
print(cline.decode())
|
|
running = False
|
|
init += 1
|
|
|
|
if out1.returncode:
|
|
print('Something went wrong:', out1.returncode)
|
|
|
|
|
|
def task2():
|
|
print("run task 2")
|
|
run_t2 = ["start", os.path.join(sys.exec_prefix, 'pythonw'), "serial_plot.pyw"]
|
|
out2 = subprocess.Popen(run_t2, shell=True, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
running = True
|
|
init = 0
|
|
while running:
|
|
cline = out2.stdout.readline()
|
|
print(cline.decode())
|
|
if not cline and init > 0:
|
|
print(cline.decode())
|
|
running = False
|
|
init += 1
|
|
|
|
if out2.returncode:
|
|
print('Something went wrong:', out2.returncode)
|
|
|
|
|
|
def run(e):
|
|
file = main()
|
|
save_diag = f"File has saved as {os.path.split(file)[1]} under {os.path.split(file)[0]} directory!\n"
|
|
t1 = Thread(target=task1, args=())
|
|
t1.setDaemon(True)
|
|
t1.start()
|
|
|
|
t2 = Thread(target=task2, args=())
|
|
t2.setDaemon(True)
|
|
t2.start()
|
|
if not frame.show_msg.GetValue():
|
|
frame.Close()
|
|
wx.MessageBox(save_diag, "Info", style=wx.ICON_INFORMATION)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
app = wx.App()
|
|
frame = Frame(None)
|
|
get_devices()
|
|
frame.SetTitle("Cease your resistance! - a0.0.0")
|
|
frame.btLaunch.Bind(wx.EVT_BUTTON, run)
|
|
frame.Show()
|
|
frame.m_textCtrl26.SetValue("480")
|
|
app.MainLoop()
|