eive/Helmholtz_Test_Bench
16
0
Fork 0
forked from zietzm/Helmholtz_Test_Bench
Code Issues Pull Requests Projects Releases 1 Wiki Activity

Fixed csv_threading.py

Browse Source
This commit is contained in:
Leon Teichroeb
2021-09-28 17:46:09 +02:00
parent 4bb6536a84
commit d02bde9631
3 changed files with 152 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/csv_threading.py
+90 -72
View File
@@ -1,16 +1,16 @@
# tThis file contains code for executing a sequence of magnetic fields from a csv file.
# This file contains code for executing a sequence of magnetic fields from a csv file.
# To do this without crashing the UI it has to run in a separate thread using the threading module.
# ToDo!: apparently max. 1 thread can access PSU --> continuous update + csv thread crashes program. Find solution
# import packages:
import time
from io import StringIO
import pandas
import numpy as np
from threading import *
from tkinter import messagebox
import matplotlib.pyplot as plt
# import other project files:
from src.exceptions import DeviceBusy, DeviceAccessError
from src.utility import ui_print
import src.user_interface as ui
import src.globals as g
@@ -20,114 +20,133 @@ class ExecCSVThread(Thread):
# main class for executing a CSV sequence
# it inherits the threading.Thread class, enabling sequence execution in a separate thread
def __init__(self, array, parent, controller, logging_enabled):
def __init__(self, array, parent, controller):
Thread.__init__(self)
self.array = array # numpy array containing data from csv to be executed
self.parent = parent # object from which this class is called, here the ExecuteCSVMode object of the UI
self.controller = controller # object on which mainloop() is running, usually the main UI window
self.logging_enabled = logging_enabled
self.__stop_event = Event() # event which can be set to stop the thread execution if needed
# Acquire cage device. This resource will only be released after the thread is ended.
try:
self.cage_dev = g.CAGE_DEVICE.request_proxy()
except DeviceBusy:
raise DeviceAccessError("Failed to acquire coil control. Required for ambient field calibration.")
self._stop_event = Event() # event which can be set to stop the thread execution if needed
def run(self): # called to start the execution of the thread
ui_print("\nStarting Sequence Execution...")
self.execute_sequence(self.array, 0.1, self.parent, self.controller) # run sequence
# when the sequence has ended, reset buttons on the UI:
if not g.exitFlag: # main window is open
self.parent.select_file_button["state"] = "normal"
self.parent.execute_button["state"] = "normal"
self.parent.stop_button["state"] = "disabled"
self.parent.reinit_button["state"] = "normal"
try:
self.execute_sequence(self.array, 0.1, self.parent, self.controller) # run sequence
finally:
self.cage_dev.idle() # set currents and voltages to 0, set arduino pins to low
# Release the proxy so other components can use it
self.cage_dev.close()
# when the sequence has ended, reset buttons on the UI:
if not g.exitFlag: # main window is open
self.parent.select_file_button["state"] = "normal"
self.parent.execute_button["state"] = "normal"
self.parent.stop_button["state"] = "disabled"
self.parent.reinit_button["state"] = "normal"
# setup ability to interrupt thread (https://stackoverflow.com/questions/323972/is-there-any-way-to-kill-a-thread)
def stop(self): # stop thread execution, can be called from another thread to kill this one
self.__stop_event.set()
self._stop_event.set()
@property
def stopped(self): # returns true if the thread has been stopped, used to check if a run should continue
return self.__stop_event.is_set()
return self._stop_event.is_set()
def execute_sequence(self, array, delay, parent, controller):
# main execution method of the class
# runs through array with times and desired fields and commands test bench accordingly
# array format: [time (s), xField (T), yField (T), zField (T)]
func.power_down_all() # sets outputs on PSUs to 0 and Arduino pins to LOW before starting
self.cage_dev.idle() # sets outputs on PSUs to 0 and Arduino pins to LOW before starting
t_zero = time.time() # set reference time for start of run
# Check if everything is properly connected:
all_connected = func.devices_ok(parent.xy_override.get(), parent.z_override.get(),
parent.arduino_override.get())
all_connected = (parent.xy_override.get() or g.CAGE_DEVICE.psu1 is not None) and\
(parent.z_override.get() or g.CAGE_DEVICE.psu2 is not None) and\
(parent.arduino_override.get() or g.CAGE_DEVICE.arduino is not None)
# True or False depending on devices status, checks for some devices may be overridden by user
if not all_connected:
ui_print("Required devices are not present, sequence aborted.")
messagebox.showwarning("Device Error!", "Required devices are not present, sequence aborted.")
return
i = 0 # index of the current array row
while i < len(array) and all_connected and not self.stopped() and not g.exitFlag:
while i < len(array):
if self.stopped or g.exitFlag:
# Interrupt sequence
ui_print("Sequence interrupted, powering down channels.")
# Channels powered down in run function
return
# while array is not finished, devices are connected, user has not cancelled and application is running
t = time.time() - t_zero # get time relative to start of run
if t >= array[i, 0]: # time for this row has come
g.threadLock.acquire() # execute all lines until threadLock.release() before going back to main thread
target_t = array[i, 0] # Target execution time of data point
if t >= target_t: # time for this row has come
field_vec = array[i, 1:4] # extract desired field vector
ui_print("[{:5.3f}s] B=[{:.1f}, {:.1f}, {:.1f}]\u03BCT for t={:.2f}s".format(t,
field_vec[0] * 1e6,
field_vec[1] * 1e6,
field_vec[2] * 1e6,
target_t))
self.cage_dev.set_field_compensated(field_vec) # send field vector to test bench
# check if everything is still connected before sending commands:
all_connected = func.devices_ok(parent.xy_override.get(), parent.z_override.get(),
parent.arduino_override.get())
if all_connected:
field_vec = array[i, 1:4] # extract desired field vector
ui_print("%0.5f s: t = %0.2f s, target field vector ="
% (time.time() - t_zero, array[i, 0]), field_vec * 1e6, "\u03BCT")
func.set_field(field_vec) # send field vector to test bench
# log change to the log file if user has selected event logging in the Configure Logging window
logger = controller.pages[ui.ConfigureLogging] # get object of logging configurator
if logger.event_logging: # data should be logged when test bench is commanded
logger.log_datapoint() # log data
# log change to the log file if user has selected event logging in the Configure Logging window
logger = controller.pages[ui.ConfigureLogging] # get object of logging configurator
if logger.event_logging: # data should be logged when test bench is commanded
logger.log_datapoint() # log data
i = i + 1 # next row
i = i + 1 # next row
g.threadLock.release() # allow going back to main thread now
elif t <= array[i, 0] - delay - 0.02: # is there enough time to sleep before the next row?
elif t <= target_t - delay - 0.02: # is there enough time to sleep before the next row?
time.sleep(delay) # sleep to give other threads time to run
if not self.stopped() and not g.exitFlag and all_connected: # sequence ended without interruption
ui_print("Sequence executed, powering down channels.")
elif all_connected: # interrupted by user
ui_print("Sequence cancelled, powering down channels.")
elif not all_connected: # interrupted by device error
ui_print("Error with at least one device, sequence aborted.")
messagebox.showwarning("Device Error!", "Error with at least one device, sequence aborted.")
else: # if this happens there is a mistake in the logic above, it really should not
# tell the user something weird happened:
ui_print("Encountered unexpected sequence end state:"
"\nThread Stopped:", self.stopped(), ", Application Closed:", g.exitFlag,
", Devices connected:", all_connected)
messagebox.showwarning("Unexpected state",
"Encountered unexpected sequence end state, see console output for details.")
func.power_down_all() # set currents and voltages to 0, set arduino pins to low
ui_print("Sequence executed, powering down channels.")
def read_csv_to_array(filepath): # convert a given csv file to a numpy array
# csv format: time (s); xField (T); yField (T); zField (T) (german excel)
# decimal commas
file = pandas.read_csv(filepath, sep=';', decimal=',', header=0) # read csv file without column headers
array = file.to_numpy() # convert csv to array
return array
# decimal or period commas. Do not use these characters as a thousands seperator!
with open(filepath, 'r') as csv_file:
# Normalize seperators
csv_string = csv_file.read().replace(',', '.')
# read csv file without column headers
file = pandas.read_csv(StringIO(csv_string), sep=';', decimal='.', header=0)
array = file.to_numpy() # convert csv to array
return array
def check_array_ok(array):
# check if any magnetic fields in an array exceed the test bench limits and if so display a warning message
values_ok = True
"""Checks if values are within limits, and if not shows a warning message."""
# check if any magnetic fields in an array exceed the test bench limits
warnings = []
for i in [0, 1, 2]: # go through axes/columns
max_val = g.AXES[i].max_comp_field[1] # get limits the test bench can do
min_val = g.AXES[i].max_comp_field[0]
data = array[:, i + 1] # extract data for this axis from array
# noinspection PyTypeChecker
if any(data > max_val) or any(data < min_val): # if any datapoint is out of bounds
values_ok = False
if not values_ok: # show warning pop-up if values are exceeding limits
messagebox.showwarning("Value Limits Warning!", "Found field values exceeding limits of test bench."
"\nSee plot and check values in csv.")
# get limits the test bench can do
min_val, max_val = g.CAGE_DEVICE.axes[i].max_comp_field
for row_idx in range(array.shape[0]):
data_point = array[row_idx, i + 1] # extract data for this axis from array
if data_point > max_val or data_point < min_val:
# Out of bounds
warnings.append({'row': row_idx+1, 'axis': g.AXIS_NAMES[i]})
# show warning pop-up if values are exceeding limits
nr_warnings = len(warnings)
if nr_warnings > 0:
warning_msg = "Found field values exceeding limits of test bench.\n"
# Only print the first three warnings
for i in range(min(nr_warnings, 3)):
warning_msg += "[Line {}] {} exceeds limits.\n".format(warnings[i]['row'], warnings[i]['axis'])
if nr_warnings > 3:
warning_msg += "And {} more...".format(nr_warnings - 3)
# Show all warnings collectively
messagebox.showwarning("Value Limits Warning!", warning_msg)
def plot_field_sequence(array, width, height): # create plot of fixed size (pixels) from array
@@ -156,8 +175,7 @@ def plot_field_sequence(array, width, height): # create plot of fixed size (pix
t = new_array[:, 0] # extract time column
for i in [0, 1, 2]: # go through all three axes
data = new_array[:, i + 1] * 1e6 # extract field column of this axis and convert to microtesla
max_val = g.AXES[i].max_comp_field[1] * 1e6 # get limits of achievable field
min_val = g.AXES[i].max_comp_field[0] * 1e6
min_val, max_val = g.CAGE_DEVICE.axes[i].max_comp_field * 1e6 # get limits of achievable field
plot = axes[i] # get appropriate subplot
plot.plot(t, data, linestyle='solid', marker='.') # plot data