import time import traceback from threading import RLock, Thread, Event from tkinter import messagebox import numpy as np from src.arduino_device import ArduinoDevice from src.psu_device import PSUDevicePS2000B, PSUDeviceQL355TP from src.utility import ui_print from src.exceptions import DeviceBusy, ProxyNotOwnedException import src.config_handling as config_handling import src.globals as g class HelmholtzCageDevice: """This is the central object for controlling all the test bench related HW. This way, access can be synchronized and exclusive to a single controller at once. This device always exists, irrespective of which devices are actually connected. Only the request_proxy, shutdown and destroy methods should be used !!! Provides subscriber interface for periodic status information. Provides proxy model to control access.""" # This class is thread safe POLLING_INTERVAL = 1 # Seconds between polling the device state def __init__(self): # Indicates all the threads should be joined self._stop_flag = Event() # --- POLLING SUBSCRIBERS --- # This is a list of object callbacks interested in receiving device status updates. # This will primarily include the front-end which wants to update its display data. # The callback functions should accept a dict as an argument of the form {'arduino':, 'axes':[{}, {}, {}]} self._subscribers = [] # --- COMMAND QUEUEING --- self.command_lock = RLock() # Indicates to the command executing thread that a new command has arrived for execution self.new_command_flag = Event() # Contains the next command to be executed self.command = None # --- PROXY MODEL SETUP --- # Locks all proxy requesting and releasing access self.proxy_lock = RLock() # Contains the id of the current in control proxy self.proxy_id = None # Contains the current context manager proxy. All other proxy instances are managed externally self.context_manager_proxy = None # --- STATUS VARIABLES --- # These are used to store information about our last command to provide status updates asynchronously self.target_field = [0, 0, 0] self.target_field_raw = [0, 0, 0] self.target_current = [0, 0, 0] # --- DEVICE SETUP --- # Must be acquired to access any hardware self.hardware_lock = RLock() # Com ports self.com_port_psu1 = None self.com_port_psu2 = None # PSU object used self.psu_type = None # Hardware object variables self.arduino = None self.psu1 = None self.psu2 = None self.connect_hardware() # --- AXIS CONFIGURATION --- # This is also hardware related, but in a separate section to keep it clean. # Get all settings from the config file self.axes = [] # Loop over axes for i in range(3): # The axes talks to the HW objects (Arduino, PSU) referenced in this object self.axes.append(Axis(i, self)) # Zero and activate channels. This is a sort of "armed" state so that we can send commands later self.idle() # --- HW COMMUNICATION THREAD --- self._cmd_exec_thread = Thread(target=self._cmd_exec_thread_method) self._cmd_exec_thread.start() self._hw_poll_thread = Thread(target=self._hw_poll_thread_method) self._hw_poll_thread.start() # TODO: Move to proxy def reconnect_hardware(self): self.shutdown() self.connect_hardware() self.idle() # TODO: Move to proxy def connect_hardware(self): """Connects devices. Does not check if they are already connected: Remember to call shutdown first""" with self.hardware_lock: # All devices are usable if the object exists. None indicates # the device is not connected/not working properly # Arduino setup try: self.arduino = ArduinoDevice() except Exception as e: self.arduino = None # show error messages to alert user ui_print("Arduino setup failed:", e) # PSU setup self.com_port_psu1 = config_handling.read_from_config("Supplies", "xy_port", config_handling.CONFIG_OBJECT) self.com_port_psu2 = config_handling.read_from_config("Supplies", "z_port", config_handling.CONFIG_OBJECT) psu_type_string = config_handling.read_from_config("Supplies", "supply_model", config_handling.CONFIG_OBJECT) if psu_type_string == "ps2000b": self.psu_type = PSUDevicePS2000B elif psu_type_string == "ql355tp": self.psu_type = PSUDeviceQL355TP else: raise Exception("Invalid psu model: {}".format(self.psu_type)) # psu1: controls xy axis try: self.psu1 = self.psu_type(self.com_port_psu1) except Exception as e: self.psu1 = None ui_print("Error creating PSU device:\n{}".format(e)) # psu2: controls z axis try: self.psu2 = self.psu_type(self.com_port_psu2) except Exception as e: self.psu2 = None ui_print("Error creating PSU device:\n{}".format(e)) def idle(self): """ Zero and activate channels """ if self.psu1 is not None: self.psu1.idle() if self.psu2 is not None: self.psu2.idle() if self.arduino is not None: self.arduino.idle() # Since these actions are not handled by the axes objects, also make sure to update their target field status for axis in self.axes: axis.target_current = 0 def request_proxy(self): """Returns a new HelmholtzCageProxy or None, depending on if access is available""" with self.proxy_lock: if not self.proxy_id: # The interface is available, return a new proxy object new_proxy = HelmholtzCageProxy(self) self.proxy_id = id(new_proxy) return new_proxy else: # The interface is occupied, the caller must tolerate that the request failed. raise DeviceBusy def __enter__(self): """Enables: with g.CAGE_DEVICE as dev:""" self.context_manager_proxy = self.request_proxy() return self.context_manager_proxy def release_proxy(self, proxy_obj): """Releases the proxy to free access for other controllers. Should only be called when proxy is destroyed""" if self.proxy_valid(proxy_obj): # This only frees the interface if it really was the active proxy self.proxy_id = None # Otherwise do nothing, this case requires no behaviour def __exit__(self, *args): """Enables: with g.CAGE_DEVICE as dev:""" self.release_proxy(self.context_manager_proxy) def proxy_valid(self, proxy_obj): """Returns True if the proxy currently owns the device.""" with self.proxy_lock: return id(proxy_obj) == self.proxy_id def subscribe_status_updates(self, callback): # List containing all interested subscribers. # We won't check if a callback is added twice. Not our responsibility self._subscribers.append(callback) def queue_command(self, proxy_obj, command): """ Queues a dict for immediate execution containing the command for the cage as a whole. Since the newest command should always be run, it is not a real queue (just a variable)""" with self.proxy_lock: if id(proxy_obj) != self.proxy_id: raise ProxyNotOwnedException() with self.command_lock: # Overwrite any command that was queued but not yet executed. We now only care about the newer command self.command = command self.new_command_flag.set() def _cmd_exec_thread_method(self): """This method forms the main thread for hardware command execution.""" while not self._stop_flag.is_set(): self.new_command_flag.wait() self.new_command_flag.clear() # Avoid blocking the buffer while we are executing. with self.command_lock: # Dicts and lists are mutable so must be (deep)copied command_buffer = HelmholtzCageDevice._copy_command(self.command) self.command = None # Processed commands are removed from "buffer" if command_buffer: # Unpack command into "action" and argument command_string = command_buffer['command'] command_arg = command_buffer['arg'] # Check which command and delegate to responsible function if command_string == "set_signed_currents": self._set_signed_currents(command_arg) elif command_string == "set_field_raw": self._set_field_raw(command_arg) elif command_string == "set_field_compensated": self._set_field_compensated(command_arg) elif command_string == 'idle': self.idle() else: raise Exception("Command unknown!") def _hw_poll_thread_method(self): """This method forms the main thread for hardware command execution.""" while True: # We will have to check if we passed this statement due to a stop flag or due to the polling interval stop_flag_set = self._stop_flag.wait(timeout=self.POLLING_INTERVAL) if stop_flag_set: return ard_conn = self.arduino is not None status_data = {'axes': [], 'arduino_connected': ard_conn} with self.hardware_lock: # This polls all three axes at once for axis in self.axes: status_data['axes'].append(axis.get_status_dict()) # Distribute status data to all interested subscribers for subscriber in self._subscribers: subscriber(status_data) @staticmethod def _copy_command(command): # PyCharm has an issue with the deepcopy tool, so just handle the copying manually. if command is None: return command try: return {'command': command['command'], 'arg': command['arg'].copy()} except AttributeError: # AttributeError: '---' object has no attribute 'copy' # Should be immutable. Otherwise we have a problem return {'command': command['command'], 'arg': command['arg']} def _set_field_raw(self, arg): for axis, field in zip(self.axes, arg): with self.hardware_lock: axis.set_field_raw(field) def _set_field_compensated(self, arg): for axis, field in zip(self.axes, arg): with self.hardware_lock: axis.set_field_compensated(field) def _set_signed_currents(self, arg): """Sets the currents in the array arg in the respective coils x->y->z. This function imposes safety limits by clamping the current when beyond the maximum.""" # One pass for every axis for axis, current in zip(self.axes, arg): # Talk to hardware with self.hardware_lock: try: axis.set_signed_current(current) except Exception as e: ui_print("Error {}: Unexpected error occured:\n{}".format(axis.name, e)) traceback.print_exc() def get_psu_for_axis(self, axis_index): """Determine which channel of which psu is required""" # TODO: This kind of stuff belongs in the config and should not be hardcoded if axis_index == 0 or axis_index == 1: psu = self.psu1 channel = self.psu_type.valid_channels()[axis_index] port = self.com_port_psu1 else: psu = self.psu2 channel = self.psu_type.valid_channels()[0] port = self.com_port_psu2 return psu, channel, port def destroy(self): """The object cannot be recovered after calling destroy""" # Send signals to kill threads: # TODO: Handle timeout behaviour self._stop_flag.set() # _cmd_exec_thread: with self.command_lock: self.command = None self.new_command_flag.set() # Causes the thread to unblock self._cmd_exec_thread.join(timeout=2) # _hw_poll_thread: # This thread is stopped just by setting the _stop_flag self._hw_poll_thread.join(timeout=2) # Shutdown the hardware msg = self.shutdown() messagebox.showinfo("Program ended", msg) # Show a unified pop-up with how the shutdown on each device went def shutdown(self): """ Shuts down the hardware. This special command overrides the currently active proxy.""" # This waiting period is not easily removed without resulting in unexpected behaviour with self.hardware_lock: ui_print("\nAttempting to safely shut down all devices. Check equipment to confirm.") # start writing string to later show how shutdown on all devices went in a single info pop-up: message = "Tried to shut down all devices. Check equipment to confirm." # Shutdown XY PSU if self.psu1 is not None: try: self.psu1.shutdown() self.psu1.destroy() self.psu1 = None except Exception as e: ui_print("Error while deactivating XY PSU:", e) # print the problem in the console message += "\nError while deactivating XY PSU: %s" % e # append status to the message to show later else: # device was successfully deactivated ui_print("XY PSU deactivated.") message += "\nXY PSU deactivated." # append message to show later else: # the device was not connected before # tell user there was no need/no possibility to deactivate: ui_print("XY PSU not connected, can't deactivate.") message += "\nXY PSU not connected, can't deactivate." # Shutdown Z PSU if self.psu2 is not None: try: self.psu2.shutdown() self.psu2.destroy() self.psu2 = None except Exception as e: ui_print("Error while deactivating Z PSU:", e) # print the problem in the console message += "\nError while deactivating Z PSU: %s" % e # append status to the message to show later else: # device was successfully deactivated ui_print("Z PSU deactivated.") message += "\nZ PSU deactivated." # append message to show later else: # the device was not connected before # tell user there was no need/no possibility to deactivate: ui_print("Z PSU not connected, can't deactivate.") message += "\nZ PSU not connected, can't deactivate." # Shut down arduino: if self.arduino is not None: try: self.arduino.shutdown() self.arduino.close() self.arduino = None except Exception as e: ui_print("Error while deactivating Arduino:", e) # print the problem in the console message += "\nError while deactivating Arduino: %s" % e # append status to the message to show later else: # device was successfully deactivated ui_print("Arduino deactivated.") message += "\nArduino deactivated." # append message to show later else: # the device was not connected before # tell user there was no need/no possibility to deactivate: ui_print("Arduino not connected, can't deactivate.") message += "\nArduino not connected, can't deactivate." return message class Axis: def __init__(self, axis_idx, cage_dev): """ This class is an adapter to axis-specific and non-specific (e.g. psu status) information. Most attributes are supplied as properties, which query the parent component of that attribute/info. It also wraps device calls. """ self.idx = axis_idx self.cage_dev = cage_dev self.name = g.AXIS_NAMES[axis_idx] # Also used in some indexing operations, like in the config file # Create other axis properties using config parameters self.coil_const = float(config_handling.read_from_config(self.name, "coil_const", config_handling.CONFIG_OBJECT)) self.ambient_field = float(config_handling.read_from_config(self.name, "ambient_field", config_handling.CONFIG_OBJECT)) self.resistance = float(config_handling.read_from_config(self.name, "resistance", config_handling.CONFIG_OBJECT)) self.max_volts = float(config_handling.read_from_config(self.name, "max_volts", config_handling.CONFIG_OBJECT)) self.max_amps = float(config_handling.read_from_config(self.name, "max_amps", config_handling.CONFIG_OBJECT)) # State variables self.target_current = 0 self.polarity = False def set_field_raw(self, field): self.set_signed_current(field / self.coil_const) def set_field_compensated(self, field): self.set_field_raw(field - self.ambient_field) def set_signed_current(self, current): """Sets current on axis""" # Check current limits if abs(current) <= self.max_amps: safe_current = current else: safe_current = self.max_amps ui_print("Warning {}: Attempted to exceed current limit".format(self.name)) # Update state variables to be queried. This should be set even if it is only a "virtual" action with no # connected devices. self.target_current = safe_current if not self.arduino: ui_print("Warning {}: Cannot set field/current without Arduino".format(self.name)) return if not self.psu: ui_print("Warning {}: Cannot set field/current without PSU".format(self.name)) return # TODO: Check for exceptions # Set polarity on Arduino if safe_current < 0: # Reverse polarity self.polarity = True # Track the state self.arduino.set_axis_polarity(self.idx, True) else: # Positive polarity (default case) self.polarity = False # Track the state self.arduino.set_axis_polarity(self.idx, False) # determine voltage limit to be set on PSU, must be high enough to not limit the current: # min. 8V, max. max_volts, in-between as needed with current value (+margin to not limit current) voltage_limit = min(max(1.3 * abs(safe_current) * self.resistance, 8), self.max_volts) # limit voltage # Set voltages and currents. Outputs should already be active from initializer. self.psu.set_current(self.channel, abs(safe_current)) self.psu.set_voltage(self.channel, voltage_limit) @property def arduino(self): return self.cage_dev.arduino @property def com_port(self): _, _, port = self.cage_dev.get_psu_for_axis(self.idx) return port @property def psu(self): psu, _, _ = self.cage_dev.get_psu_for_axis(self.idx) return psu @property def channel(self): _, channel, _ = self.cage_dev.get_psu_for_axis(self.idx) return channel @property def target_field(self): return self.target_current * self.coil_const + self.ambient_field @property def target_field_raw(self): return self.target_current * self.coil_const @property def connected(self): return self.psu is not None and self.arduino is not None @property def max_field(self): max_field_magnitude = self.max_amps * self.coil_const return np.array([-max_field_magnitude, max_field_magnitude]) @property def max_comp_field(self): max_field_magnitude = self.max_amps * self.coil_const return np.array([self.ambient_field - max_field_magnitude, self.ambient_field + max_field_magnitude]) @property def arduino_connected(self): return self.arduino is not None @property def psu_connected(self): return self.psu is not None def get_status_dict(self): """Dict containing all data from this model to pass to the front-end. Some data is only available through this interface, since it also polls the hardware for current set-points""" # This is a slow operation, watch out! if self.psu: status = self.psu.poll_channel_state(self.channel) else: status = {} if self.arduino: status['polarity'] = self.polarity status['connected'] = self.connected status['port'] = self.com_port status['channel'] = self.channel status['target_field_raw'] = self.target_field_raw status['target_field'] = self.target_field status['target_current'] = self.target_current return status class HelmholtzCageProxy: """ Proxy for the HelmholtzCageDevice. This is the only way the application should communicate with the HelmholtzCageDevice object""" def __init__(self, cage_device): self.cage_device = cage_device def set_signed_currents(self, vector): self.cage_device.queue_command(self, {'command': 'set_signed_currents', 'arg': vector}) def set_field_raw(self, vector): self.cage_device.queue_command(self, {'command': 'set_field_raw', 'arg': vector}) def set_field_compensated(self, vector): self.cage_device.queue_command(self, {'command': 'set_field_compensated', 'arg': vector}) def idle(self): """Puts the helmholtz cage into an idle state with zeroed fields""" self.cage_device.queue_command(self, {'command': 'idle', 'arg': None}) def close(self): self.cage_device.release_proxy(self) def __del__(self): # This is a fallback method and should not be relied on. Call 'close' manually if self.cage_device.proxy_valid(self): self.cage_device.release_proxy(self) ui_print("Warning: Proxy implicitly released. Use close() instead.") def value_in_limits(axis, key, value): """Check if value is within safe limits (set in globals.py)""" # axis is string with axis name, e.g. "X-Axis" # key specifies which value to check, e.g. current max_value = g.default_arrays[key][1][g.AXIS_NAMES.index(axis)] # get max value from dictionary in globals.py min_value = g.default_arrays[key][2][g.AXIS_NAMES.index(axis)] # get min value from dictionary in globals.py if float(value) > float(max_value): # value is too high return 'HIGH' elif float(value) < float(min_value): # value is too low return 'LOW' else: # value is within limits return 'OK'