diff --git a/software/control/gui_hcs.py b/software/control/gui_hcs.py index 388e74ad8..511aa0ae6 100644 --- a/software/control/gui_hcs.py +++ b/software/control/gui_hcs.py @@ -1,9 +1,8 @@ # set QT_API environment variable import os -import sys -import time os.environ["QT_API"] = "pyqt5" -import qtpy +import serial +import time # qt libraries from qtpy.QtCore import * @@ -14,37 +13,40 @@ # app specific libraries import control.widgets as widgets -import serial +import pyqtgraph.dockarea as dock +import squid.logging + +log = squid.logging.get_logger(__name__) if CAMERA_TYPE == "Toupcam": try: import control.camera_toupcam as camera except: - print("Problem importing Toupcam, defaulting to default camera") + log.warning("Problem importing Toupcam, defaulting to default camera") import control.camera as camera elif CAMERA_TYPE == "FLIR": try: import control.camera_flir as camera except: - print("Problem importing FLIR camera, defaulting to default camera") + log.warning("Problem importing FLIR camera, defaulting to default camera") import control.camera as camera elif CAMERA_TYPE == "Hamamatsu": try: import control.camera_hamamatsu as camera except: - print("Problem importing Hamamatsu camera, defaulting to default camera") + log.warning("Problem importing Hamamatsu camera, defaulting to default camera") import control.camera as camera elif CAMERA_TYPE == "iDS": try: import control.camera_ids as camera except: - print("Problem importing iDS camera, defaulting to default camera") + log.warning("Problem importing iDS camera, defaulting to default camera") import control.camera as camera elif CAMERA_TYPE == "Tucsen": try: import control.camera_tucsen as camera except: - print("Problem importing Tucsen camera, defaulting to default camera") + log.warning("Problem importing Tucsen camera, defaulting to default camera") import control.camera as camera else: import control.camera as camera @@ -53,13 +55,13 @@ try: import control.camera_toupcam as camera_fc except: - print("Problem importing Toupcam for focus, defaulting to default camera") + log.warning("Problem importing Toupcam for focus, defaulting to default camera") import control.camera as camera_fc elif FOCUS_CAMERA_TYPE == "FLIR": try: import control.camera_flir as camera_fc except: - print("Problem importing FLIR camera for focus, defaulting to default camera") + log.warning("Problem importing FLIR camera for focus, defaulting to default camera") import control.camera as camera_fc else: import control.camera as camera_fc @@ -78,15 +80,10 @@ if SUPPORT_LASER_AUTOFOCUS: import control.core_displacement_measurement as core_displacement_measurement -import pyqtgraph.dockarea as dock -import serial -import time - SINGLE_WINDOW = True # set to False if use separate windows for display and control class OctopiGUI(QMainWindow): - fps_software_trigger = 100 def __init__(self, is_simulation=False, performance_mode=False, *args, **kwargs): @@ -125,11 +122,9 @@ def loadObjects(self, is_simulation): else: try: self.loadHardwareObjects() - except Exception as e: - print("\n---- !! ERROR CONNECTING TO HARDWARE !! ----") - print(e) - print("Falling back to simulation mode\n") - self.loadSimulationObjects() + except Exception: + log.error("---- !! ERROR CONNECTING TO HARDWARE !! ----", stack_info=True, exc_info=True) + raise # Common object initialization self.objectiveStore = core.ObjectiveStore(parent=self) @@ -164,6 +159,7 @@ def loadObjects(self, is_simulation): self.laserAutofocusController = core.LaserAutofocusController(self.microcontroller,self.camera_focus,self.liveController_focus_camera,self.navigationController,has_two_interfaces=HAS_TWO_INTERFACES,use_glass_top=USE_GLASS_TOP,look_for_cache=False) def loadSimulationObjects(self): + log.debug("Loading simulated hardware objects...") # Initialize simulation objects if ENABLE_SPINNING_DISK_CONFOCAL: self.xlight = serial_peripherals.XLight_Simulation() @@ -187,15 +183,17 @@ def loadHardwareObjects(self): if ENABLE_SPINNING_DISK_CONFOCAL: try: self.xlight = serial_peripherals.XLight(XLIGHT_SERIAL_NUMBER, XLIGHT_SLEEP_TIME_FOR_WHEEL) - except Exception as e: - raise Exception(f"Error initializing Spinning Disk Confocal: {e}") + except Exception: + log.error("Error initializing Spinning Disk Confocal") + raise if ENABLE_NL5: try: import control.NL5 as NL5 self.nl5 = NL5.NL5() - except Exception as e: - raise Exception(f"Error initializing NL5: {e}") + except Exception: + log.error("Error initializing NL5") + raise if ENABLE_CELLX: try: @@ -203,8 +201,9 @@ def loadHardwareObjects(self): for channel in [1,2,3,4]: self.cellx.set_modulation(channel, CELLX_MODULATION) self.cellx.turn_on(channel) - except Exception as e: - raise Exception(f"Error initializing CellX: {e}") + except Exception: + log.error("Error initializing CellX") + raise if USE_LDI_SERIAL_CONTROL: try: @@ -212,8 +211,9 @@ def loadHardwareObjects(self): self.ldi.run() self.ldi.set_intensity_mode(LDI_INTENSITY_MODE) self.ldi.set_shutter_mode(LDI_SHUTTER_MODE) - except Exception as e: - raise Exception(f"Error initializing LDI: {e}") + except Exception: + log.error("Error initializing LDI") + raise if SUPPORT_LASER_AUTOFOCUS: try: @@ -221,39 +221,45 @@ def loadHardwareObjects(self): self.camera_focus = camera_fc.Camera(sn=sn_camera_focus) self.camera_focus.open() self.camera_focus.set_pixel_format('MONO8') - except Exception as e: - raise Exception(f"Error initializing Laser Autofocus Camera: {e}") + except Exception: + log.error(f"Error initializing Laser Autofocus Camera") + raise try: sn_camera_main = camera.get_sn_by_model(MAIN_CAMERA_MODEL) self.camera = camera.Camera(sn=sn_camera_main, rotate_image_angle=ROTATE_IMAGE_ANGLE, flip_image=FLIP_IMAGE) self.camera.open() self.camera.set_pixel_format(DEFAULT_PIXEL_FORMAT) - except Exception as e: - raise Exception(f"Error initializing Main Camera: {e}") + except Exception: + log.error("Error initializing Main Camera") + raise if USE_ZABER_EMISSION_FILTER_WHEEL: try: self.emission_filter_wheel = serial_peripherals.FilterController(FILTER_CONTROLLER_SERIAL_NUMBER, 115200, 8, serial.PARITY_NONE, serial.STOPBITS_ONE) - except Exception as e: - raise Exception(f"Error initializing Zaber Emission Filter Wheel: {e}") + except Exception: + log.error("Error initializing Zaber Emission Filter Wheel") + raise if USE_OPTOSPIN_EMISSION_FILTER_WHEEL: try: self.emission_filter_wheel = serial_peripherals.Optospin(SN=FILTER_CONTROLLER_SERIAL_NUMBER) - except Exception as e: - raise Exception(f"Error initializing Optospin Emission Filter Wheel: {e}") + except Exception: + log.error("Error initializing Optospin Emission Filter Wheel") + raise if USE_PRIOR_STAGE: try: self.priorstage = PriorStage(PRIOR_STAGE_SN, parent=self) - except Exception as e: - raise Exception(f"Error initializing Prior Stage: {e}") + except Exception: + log.error("Error initializing Prior Stage") + raise try: self.microcontroller = microcontroller.Microcontroller(version=CONTROLLER_VERSION, sn=CONTROLLER_SN) - except Exception as e: - raise Exception(f"Error initializing Microcontroller: {e}") + except Exception: + log.error(f"Error initializing Microcontroller") + raise def setupHardware(self): # Setup hardware components @@ -335,7 +341,7 @@ def waitForMicrocontroller(self, timeout=None, error_message=None): while self.microcontroller.is_busy(): time.sleep(0.005) if timeout and time.time() - start_time > timeout: - print(error_message or 'Microcontroller operation timed out') + self.log.error(error_message or 'Microcontroller operation timed out') sys.exit(1) def loadWidgets(self): diff --git a/software/control/serial_peripherals.py b/software/control/serial_peripherals.py index 7c099b74b..fa109506f 100644 --- a/software/control/serial_peripherals.py +++ b/software/control/serial_peripherals.py @@ -4,6 +4,9 @@ from typing import Tuple, Optional import struct +import squid.logging +log = squid.logging.get_logger(__name__) + class SerialDevice: """ General wrapper for serial devices, with @@ -82,12 +85,12 @@ def write_and_check(self, command, expected_response, read_delay=0.1, max_attemp response = self.serial.readline().decode().strip() if print_response: - print(response) + log.info(response) # flush the input buffer while self.serial.in_waiting: if print_response: - print(self.serial.readline().decode().strip()) + log.info(self.serial.readline().decode().strip()) else: self.serial.readline().decode().strip() @@ -95,8 +98,8 @@ def write_and_check(self, command, expected_response, read_delay=0.1, max_attemp if response == expected_response: return response else: - print(response) - + log.warning(response) + # check prefix if the full response does not match if check_prefix: if response.startswith(expected_response): @@ -149,7 +152,7 @@ def set_dichroic(self, position, extraction=False): def get_dichroic(self): return self.dichroic_wheel_pos - + def set_disk_position(self, position): self.spinning_disk_pos = position return position @@ -188,6 +191,7 @@ def set_filter_slider(self,position): # no flow control class XLight: + """Wrapper for communicating with CrestOptics X-Light devices over serial""" def __init__(self, SN, sleep_time_for_wheel = 0.25): """ @@ -195,6 +199,8 @@ def __init__(self, SN, sleep_time_for_wheel = 0.25): cephla already has) for device-finding purposes. Otherwise, all XLight devices should use the same serial protocol """ + self.log = squid.logging.get_logger(self.__class__.__name__) + self.has_spinning_disk_motor = False self.has_spinning_disk_slider = False self.has_dichroic_filters_wheel = False @@ -231,17 +237,18 @@ def parse_idc_response(self, response): self.has_ttl_control = bool(config_value & 0x00001000) def print_config(self): - print("Machine Configuration:") - print(f"Spinning disk motor: {self.has_spinning_disk_motor}") - print(f"Spinning disk slider: {self.has_spinning_disk_slider}") - print(f"Dichroic filters wheel: {self.has_dichroic_filters_wheel}") - print(f"Emission filters wheel: {self.has_emission_filters_wheel}") - print(f"Excitation filters wheel: {self.has_excitation_filters_wheel}") - print(f"Illumination Iris diaphragm: {self.has_illumination_iris_diaphragm}") - print(f"Emission Iris diaphragm: {self.has_emission_iris_diaphragm}") - print(f"Dichroic filter slider: {self.has_dichroic_filter_slider}") - print(f"TTL control and combined commands subsystem: {self.has_ttl_control}") - + self.log.info(( + "Machine Configuration:\n" + f" Spinning disk motor: {self.has_spinning_disk_motor}\n", + f" Spinning disk slider: {self.has_spinning_disk_slider}\n", + f" Dichroic filters wheel: {self.has_dichroic_filters_wheel}\n", + f" Emission filters wheel: {self.has_emission_filters_wheel}\n", + f" Excitation filters wheel: {self.has_excitation_filters_wheel}\n", + f" Illumination Iris diaphragm: {self.has_illumination_iris_diaphragm}\n", + f" Emission Iris diaphragm: {self.has_emission_iris_diaphragm}\n", + f" Dichroic filter slider: {self.has_dichroic_filter_slider}\n", + f" TTL control and combined commands subsystem: {self.has_ttl_control}")) + def set_emission_filter(self,position,extraction=False,validate=True): if str(position) not in ["1","2","3","4","5","6","7","8"]: raise ValueError("Invalid emission filter wheel position!") @@ -349,14 +356,15 @@ def __init__(self, SN="00000001"): """ Provide serial number """ + self.log = squid.logging.get_logger(self.__class__.__name__) self.serial_connection = SerialDevice(SN=SN,baudrate=9600, bytesize=serial.EIGHTBITS,stopbits=serial.STOPBITS_ONE, - parity=serial.PARITY_NONE, + parity=serial.PARITY_NONE, xonxoff=False,rtscts=False,dsrdtr=False) self.serial_connection.open_ser() self.intensity_mode = 'PC' self.shutter_mode = 'PC' - + def run(self): self.serial_connection.write_and_check("run!\r","ok") @@ -373,10 +381,10 @@ def set_intensity_mode(self,mode): def set_intensity(self,channel,intensity): channel = str(channel) intensity = "{:.2f}".format(intensity) - print('set:'+channel+'='+intensity+'\r') + self.log.debug('set:'+channel+'='+intensity+'\r') self.serial_connection.write_and_check('set:'+channel+'='+intensity+'\r',"ok") - print('active channel: ' + str(self.active_channel)) - + self.log.debug('active channel: ' + str(self.active_channel)) + def set_shutter(self,channel,state): channel = str(channel) state = str(state) @@ -387,12 +395,12 @@ def get_shutter_state(self): def set_active_channel(self,channel): self.active_channel = channel - print('[set active channel to ' + str(channel) + ']') + self.log.debug('[set active channel to ' + str(channel) + ']') def set_active_channel_shutter(self,state): channel = str(self.active_channel) state = str(state) - print('shutter:'+channel+'='+state+'\r') + self.log.debug('shutter:'+channel+'='+state+'\r') self.serial_connection.write_and_check('shutter:'+channel+'='+state+'\r',"ok") class SciMicroscopyLEDArray: @@ -403,7 +411,7 @@ def __init__(self, SN, array_distance = 50, turn_on_delay = 0.03): """ self.serial_connection = SerialDevice(SN=SN,baudrate=115200, bytesize=serial.EIGHTBITS,stopbits=serial.STOPBITS_ONE, - parity=serial.PARITY_NONE, + parity=serial.PARITY_NONE, xonxoff=False,rtscts=False,dsrdtr=False) self.serial_connection.open_ser() self.check_about() @@ -465,6 +473,9 @@ def turn_off_illumination(self): self.clear() class CellX: + + VALID_MODULATIONS = ['INT','EXT Digital','EXT Analog','EXT Mixed'] + """Wrapper for communicating with LDI over serial""" def __init__(self, SN=""): self.serial_connection = SerialDevice(SN=SN,baudrate=115200, @@ -481,11 +492,9 @@ def turn_off(self, channel): self.serial_connection.write_and_check('SOUR'+str(channel)+':AM:STAT OFF\r','OK',read_delay=0.01,print_response=False) def set_laser_power(self, channel, power): - try: - assert power >= 1 and power <= 100 - except AssertionError as e: - print(f"AssertionError: {e}") - return + if not (power >= 1 and power <= 100): + raise ValueError(f"Power={power} not in the range 1 to 100") + if channel not in self.power.keys() or power != self.power[channel]: self.serial_connection.write_and_check('SOUR'+str(channel)+':POW:LEV:IMM:AMPL '+str(power/1000)+'\r','OK',read_delay=0.01,print_response=False) self.power[channel] = power @@ -493,11 +502,8 @@ def set_laser_power(self, channel, power): pass # power is the same def set_modulation(self, channel, modulation): - try: - assert modulation in ['INT','EXT Digital','EXT Analog','EXT Mixed'] - except AssertionError as e: - print(f"AssertionError: {e}") - return + if modulation not in CellX.VALID_MODULATIONS: + raise ValueError(f"Modulation '{modulation}' not in valid modulations: {CellX.VALID_MODULATIONS}") self.serial_connection.write_and_check('SOUR'+str(channel)+':AM:' + modulation +'\r','OK',read_delay=0.01,print_response=False) def close(self): @@ -520,22 +526,17 @@ def turn_off(self, channel): pass def set_laser_power(self, channel, power): - try: - assert power >= 1 and power <= 100 - except AssertionError as e: - print(f"AssertionError: {e}") - return + if not (power >= 1 and power <= 100): + raise ValueError(f"Power={power} not in the range 1 to 100") + if channel not in self.power.keys() or power != self.power[channel]: self.power[channel] = power else: pass # power is the same def set_modulation(self, channel, modulation): - try: - assert modulation in ['INT','EXT Digital','EXT Analog','EXT Mixed'] - except AssertionError as e: - print(f"AssertionError: {e}") - return + if modulation not in CellX.VALID_MODULATIONS: + raise ValueError(f"modulation '{modulation}' not in valid choices: {CellX.VALID_MODULATIONS}") self.serial_connection.write_and_check('SOUR'+str(channel)+'AM:' + modulation +'\r','OK',read_delay=0.01,print_response=False) def close(self): @@ -599,6 +600,7 @@ class FilterController: COMMAND_TIMEOUT = 1 # seconds def __init__(self, serial_number: str, baudrate: int, bytesize: int, parity: str, stopbits: int): + self.log = squid.logging.get_logger(self.__class__.__name__) self.current_position = 0 self.current_index = 1 self.serial = self._initialize_serial(serial_number, baudrate, bytesize, parity, stopbits) @@ -645,7 +647,7 @@ def _send_command(self, cmd: str) -> Tuple[bool, str]: self.serial.write(f"{cmd}\n".encode('utf-8')) response = self.serial.readline().decode('utf-8').strip() success, message = self._parse_response(response) - + if success: return True, message elif message.startswith('BUSY'): @@ -653,12 +655,12 @@ def _send_command(self, cmd: str) -> Tuple[bool, str]: continue else: # Log the error and retry - print(f"Command failed (attempt {attempt + 1}): {message}") + self.log.error(f"Command failed (attempt {attempt + 1}): {message}") except serial.SerialTimeoutException: - print(f"Command timed out (attempt {attempt + 1})") - + self.log.error(f"Command timed out (attempt {attempt + 1})") + time.sleep(0.5) # Wait before retrying - + raise FilterControllerError(f"Command '{cmd}' failed after {self.MAX_RETRIES} attempts") def _parse_response(self, response: str) -> Tuple[bool, str]: @@ -673,7 +675,7 @@ def _parse_response(self, response: str) -> Tuple[bool, str]: """ if not response: return False, "No response received" - + parts = response.split() if len(parts) < 4: return False, f"Invalid response format: {response}" @@ -734,14 +736,14 @@ def set_emission_filter(self, index: int, blocking: bool = True, timeout: int = TimeoutError: If the movement doesn't complete within the specified timeout (only in blocking mode). """ if index not in self.VALID_POSITIONS: - raise ValueError(f"Invalid emission filter wheel position: {position}") - + raise ValueError(f"Invalid emission filter wheel index position: {index}") + target_position = self.OFFSET_POSITION + (index - 1) * self.MICROSTEPS_PER_HOLE success, _ = self._send_command(f'/move abs {target_position}') - + if not success: raise FilterControllerError("Failed to initiate filter movement") - + if blocking: self._wait_for_position(target_position, index, timeout) else: @@ -832,6 +834,7 @@ def complete_homing_sequence(self, timeout: int = 50): class Optospin: def __init__(self, SN, baudrate=115200, timeout=1, max_retries=3, retry_delay=0.5): + self.log = squid.logging.get_logger(self.__class__.__name__) optospin_port = [p.device for p in serial.tools.list_ports.comports() if SN == p.serial_number] self.ser = serial.Serial(optospin_port[0], baudrate=baudrate, timeout=timeout) @@ -865,9 +868,9 @@ def _send_command(self, command, data=None): return None except (serial.SerialTimeoutException, Exception) as e: - print(f"Attempt {attempt + 1} failed: {str(e)}") + self.log.error(f"Attempt {attempt + 1} failed: {str(e)}") if attempt < self.max_retries - 1: - print(f"Retrying in {self.retry_delay} seconds...") + self.log.error(f"Retrying in {self.retry_delay} seconds...") time.sleep(self.retry_delay) else: raise Exception(f"Command failed after {self.max_retries} attempts: {str(e)}") diff --git a/software/main_hcs.py b/software/main_hcs.py index 3a09aaf92..c78766bd8 100644 --- a/software/main_hcs.py +++ b/software/main_hcs.py @@ -1,8 +1,10 @@ # set QT_API environment variable import argparse import glob +import logging import os os.environ["QT_API"] = "pyqt5" +import signal import sys # qt libraries @@ -32,10 +34,15 @@ def show_acq_config(cfm): parser = argparse.ArgumentParser() parser.add_argument("--simulation", help="Run the GUI with simulated hardware.", action='store_true') parser.add_argument("--performance", help="Run the GUI with minimal viewers.", action='store_true') + parser.add_argument("--verbose", help="Turn on verbose logging (DEBUG level)", action="store_true") args = parser.parse_args() log = squid.logging.get_logger("main_hcs") + if args.verbose: + log.info("Turning on debug logging.") + squid.logging.set_log_level(logging.DEBUG) + legacy_config = False cf_editor_parser = ConfigParser() config_files = glob.glob('.' + '/' + 'configuration*.ini') @@ -46,6 +53,8 @@ def show_acq_config(cfm): legacy_config = True app = QApplication([]) app.setStyle('Fusion') + # This allows shutdown via ctrl+C even after the gui has popped up. + signal.signal(signal.SIGINT, signal.SIG_DFL) win = gui.OctopiGUI(is_simulation=args.simulation, performance_mode=args.performance) diff --git a/software/squid/logging.py b/software/squid/logging.py index 55a8d8c98..6aa8ec490 100644 --- a/software/squid/logging.py +++ b/software/squid/logging.py @@ -138,4 +138,4 @@ def setup_uncaught_exception_logging(): def uncaught_exception_logger(exception_type: Type[BaseException], value: BaseException, tb: TracebackType): logger.exception("Uncaught Exception!", exc_info=value) - register_crash_handler(uncaught_exception_logger, call_existing_too=False) \ No newline at end of file + register_crash_handler(uncaught_exception_logger, call_existing_too=False)