From 5a4df7818512719e986b01cb3775fc321fa4cab9 Mon Sep 17 00:00:00 2001 From: Kevin Dean <42547789+AdvancedImagingUTSW@users.noreply.github.com> Date: Sat, 5 Oct 2024 10:25:23 -0500 Subject: [PATCH 1/5] Type hinting... --- src/navigate/model/devices/daq/ni.py | 39 ++++++++-------- src/navigate/model/devices/lasers/base.py | 4 +- src/navigate/model/devices/lasers/ni.py | 18 +++++--- src/navigate/model/microscope.py | 54 ++++++++++++++--------- 4 files changed, 66 insertions(+), 49 deletions(-) diff --git a/src/navigate/model/devices/daq/ni.py b/src/navigate/model/devices/daq/ni.py index 1be74dc9b..2ad48550f 100644 --- a/src/navigate/model/devices/daq/ni.py +++ b/src/navigate/model/devices/daq/ni.py @@ -35,6 +35,7 @@ from threading import Lock import traceback import time +from typing import Union # Third Party Imports import nidaqmx @@ -56,7 +57,7 @@ class NIDAQ(DAQBase): """NIDAQ class for Control of NI Data Acquisition Cards.""" - def __init__(self, configuration): + def __init__(self, configuration: dict) -> None: """Initialize NIDAQ class. Parameters @@ -103,16 +104,16 @@ def __init__(self, configuration): #: bool: Flag for waiting to run. self.wait_to_run_lock = Lock() - def __str__(self): + def __str__(self) -> str: """String representation of the class.""" return "NIDAQ" - def __del__(self): + def __del__(self) -> None: """Destructor.""" if self.camera_trigger_task is not None: self.stop_acquisition() - def set_external_trigger(self, external_trigger=None): + def set_external_trigger(self, external_trigger=None) -> None: """Set trigger mode. Parameters @@ -191,9 +192,10 @@ def set_external_trigger(self, external_trigger=None): task.register_done_event(None) task.register_done_event(self.restart_analog_task_callback_func(task)) + @staticmethod def wait_for_external_trigger( - self, trigger_channel, wait_internal=0.001, timeout=-1 - ): + trigger_channel: str, wait_internal=0.001, timeout=-1 + ) -> bool: """Wait for a digital external trigger. Parameters @@ -228,24 +230,23 @@ def wait_for_external_trigger( time.sleep(wait_internal) total_wait_time += wait_internal - if timeout > 0 and total_wait_time >= timeout: + if 0 < timeout <= total_wait_time: result = False break # Close the task external_trigger_task.stop() external_trigger_task.close() - return result @staticmethod - def restart_analog_task_callback_func(task): + def restart_analog_task_callback_func(task: nidaqmx.Task) -> callable: """Restart analog task callback function. Parameters ---------- task : nidaqmx.Task - Task for analog output + Analog output task. Returns ------- @@ -264,7 +265,7 @@ def callback_func(task_handle, status, callback_data): return callback_func - def create_camera_task(self, channel_key): + def create_camera_task(self, channel_key: str) -> None: """Set up the camera trigger task. TTL for triggering the camera. TTL is 4 ms in duration. @@ -309,7 +310,7 @@ def create_camera_task(self, channel_key): samps_per_chan=camera_waveform_repeat_num, ) - def create_master_trigger_task(self): + def create_master_trigger_task(self) -> None: """Set up the DO master trigger task.""" self.master_trigger_task = nidaqmx.Task() master_trigger_out_line = self.configuration["configuration"]["microscopes"][ @@ -320,7 +321,7 @@ def create_master_trigger_task(self): line_grouping=nidaqmx.constants.LineGrouping.CHAN_FOR_ALL_LINES, ) - def create_analog_output_tasks(self, channel_key): + def create_analog_output_tasks(self, channel_key: str) -> None: """Create analog output tasks for each board. Create a single analog output task for all channels per board. Most NI DAQ cards @@ -386,7 +387,7 @@ def create_analog_output_tasks(self, channel_key): ).squeeze() self.analog_output_tasks[board].write(waveforms) - def prepare_acquisition(self, channel_key): + def prepare_acquisition(self, channel_key: str) -> None: """Prepare the acquisition. Creates and configures the DAQ tasks. @@ -422,7 +423,7 @@ def prepare_acquisition(self, channel_key): # Specify ports, timing, and triggering self.set_external_trigger(self.external_trigger) - def run_acquisition(self): + def run_acquisition(self) -> None: """Run DAQ Acquisition. Run the tasks for triggering, analog and counter outputs. @@ -463,7 +464,7 @@ def run_acquisition(self): except nidaqmx.DaqError: pass - def stop_acquisition(self): + def stop_acquisition(self) -> None: """Stop Acquisition. Stop all tasks and close them. @@ -488,7 +489,7 @@ def stop_acquisition(self): self.analog_output_tasks = {} - def enable_microscope(self, microscope_name): + def enable_microscope(self, microscope_name: str) -> None: """Enable microscope. Parameters @@ -528,7 +529,7 @@ def enable_microscope(self, microscope_name): except KeyError: pass - def update_analog_task(self, board_name): + def update_analog_task(self, board_name: str) -> Union[bool, None]: """Update analog task. Parameters @@ -538,7 +539,7 @@ def update_analog_task(self, board_name): Returns ------- - bool + bool, None True if task is updated, False otherwise. """ # if there is no such analog task, diff --git a/src/navigate/model/devices/lasers/base.py b/src/navigate/model/devices/lasers/base.py index b37d56d5d..30f6b4791 100644 --- a/src/navigate/model/devices/lasers/base.py +++ b/src/navigate/model/devices/lasers/base.py @@ -57,8 +57,8 @@ def __init__(self, microscope_name, device_connection, configuration, laser_id): ---------- microscope_name : str Name of the microscope - device_connection : str - Connection string for the device + device_connection : object + Communication instance with the device. configuration : dict Configuration dictionary laser_id : int diff --git a/src/navigate/model/devices/lasers/ni.py b/src/navigate/model/devices/lasers/ni.py index a9337aa12..a69e4d272 100644 --- a/src/navigate/model/devices/lasers/ni.py +++ b/src/navigate/model/devices/lasers/ni.py @@ -54,7 +54,13 @@ class LaserNI(LaserBase): This class is used to control a laser connected to a National Instruments DAQ. """ - def __init__(self, microscope_name, device_connection, configuration, laser_id): + def __init__( + self, + microscope_name: str, + device_connection: object, + configuration: dict, + laser_id: int, + ) -> None: """Initialize the LaserNI class. Parameters @@ -65,7 +71,7 @@ def __init__(self, microscope_name, device_connection, configuration, laser_id): The device connection object. configuration : dict The device configuration. - laser_id : str + laser_id : int The laser id. """ super().__init__(microscope_name, device_connection, configuration, laser_id) @@ -130,7 +136,7 @@ def __init__(self, microscope_name, device_connection, configuration, laser_id): logger.debug(f"{str(self)} error:, {e}, {e.error_type}, {e.error_code}") print(f"{str(self)} error:, {e}, {e.error_type}, {e.error_code}") - def set_power(self, laser_intensity): + def set_power(self, laser_intensity: float) -> None: """Sets the laser power. Parameters @@ -145,7 +151,7 @@ def set_power(self, laser_intensity): except DaqError as e: logger.exception(e) - def turn_on(self): + def turn_on(self) -> None: """Turns on the laser.""" try: self.set_power(self._current_intensity) @@ -157,7 +163,7 @@ def turn_on(self): except DaqError as e: logger.exception(e) - def turn_off(self): + def turn_off(self) -> None: """Turns off the laser.""" try: tmp = self._current_intensity @@ -171,7 +177,7 @@ def turn_off(self): except DaqError as e: logger.exception(e) - def close(self): + def close(self) -> None: """Close the NI Task before exit.""" try: self.laser_ao_task.close() diff --git a/src/navigate/model/microscope.py b/src/navigate/model/microscope.py index 0631c6c95..6c519b336 100644 --- a/src/navigate/model/microscope.py +++ b/src/navigate/model/microscope.py @@ -406,12 +406,14 @@ def prepare_acquisition(self): Dictionary of all the waveforms. """ camera_info = reprlib.Repr() - camera_info.indent = '---' + camera_info.indent = "---" camera_info.maxdict = 20 camera_info = camera_info.repr( - dict(self.configuration['experiment']['CameraParameters'][ - self.microscope_name] - ) + dict( + self.configuration["experiment"]["CameraParameters"][ + self.microscope_name + ] + ) ) logger.info(f"Preparing Acquisition. Camera Parameters: {camera_info}") @@ -486,12 +488,16 @@ def end_acquisition(self): def turn_on_laser(self): """Turn on the current laser.""" - logger.info(f"Turning on laser {self.laser_wavelength[self.current_laser_index]}") + logger.info( + f"Turning on laser {self.laser_wavelength[self.current_laser_index]}" + ) self.lasers[str(self.laser_wavelength[self.current_laser_index])].turn_on() def turn_off_lasers(self): """Turn off current laser.""" - logger.info(f"Turning off laser {self.laser_wavelength[self.current_laser_index]}") + logger.info( + f"Turning off laser {self.laser_wavelength[self.current_laser_index]}" + ) self.lasers[str(self.laser_wavelength[self.current_laser_index])].turn_off() def calculate_all_waveform(self): @@ -750,7 +756,9 @@ def prepare_next_channel(self, update_daq_task_flag=True): update_focus=False, ) - def move_stage(self, pos_dict, wait_until_done=False, update_focus=True): + def move_stage( + self, pos_dict: dict, wait_until_done=False, update_focus=True + ) -> bool: """Move stage to a position. Parameters @@ -793,7 +801,7 @@ def move_stage(self, pos_dict, wait_until_done=False, update_focus=True): return success - def stop_stage(self): + def stop_stage(self) -> None: """Stop stage.""" self.ask_stage_for_position = True @@ -803,7 +811,7 @@ def stop_stage(self): self.central_focus = self.get_stage_position().get("f_pos", self.central_focus) - def get_stage_position(self): + def get_stage_position(self) -> dict: """Get stage position. Returns @@ -818,7 +826,7 @@ def get_stage_position(self): self.ask_stage_for_position = False return self.ret_pos_dict - def move_remote_focus(self, offset=None): + def move_remote_focus(self, offset=None) -> None: """Move remote focus. Parameters @@ -829,7 +837,7 @@ def move_remote_focus(self, offset=None): exposure_times, sweep_times = self.calculate_exposure_sweep_times() self.remote_focus_device.move(exposure_times, sweep_times, offset) - def update_stage_limits(self, limits_flag=True): + def update_stage_limits(self, limits_flag=True) -> None: """Update stage limits. Parameters @@ -842,7 +850,7 @@ def update_stage_limits(self, limits_flag=True): for stage, _ in self.stages_list: stage.stage_limits = limits_flag - def assemble_device_config_lists(self, device_name, device_name_dict): + def assemble_device_config_lists(self, device_name: str, device_name_dict: dict): """Assemble device config lists. Parameters @@ -892,15 +900,15 @@ def assemble_device_config_lists(self, device_name, device_name_dict): def load_and_start_devices( self, - device_name, - is_list, - device_name_list, - device_ref_name, + device_name: str, + is_list: bool, + device_name_list: list, + device_ref_name: str, device_connection, - name, - i, - plugin_devices, - ): + name: str, + i: int, + plugin_devices: dict, + ) -> None: """Load and start devices. Parameters @@ -921,6 +929,8 @@ def load_and_start_devices( Index. plugin_devices : dict Plugin Devices + + TODO: Remove uncalled parameters (device_connection, name, plugin_devices)? """ # Import start_device classes try: @@ -949,7 +959,7 @@ def load_and_start_devices( ) self.info[device_name] = device_ref_name - def terminate(self): + def terminate(self) -> None: """Close hardware explicitly.""" self.camera.close_camera() @@ -969,7 +979,7 @@ def terminate(self): print(f"Stage delete failure: {e}") pass - def run_command(self, command, *args): + def run_command(self, command: str, *args) -> None: """Run command. Parameters From e508839e9479b92707fb6b4f09f5c7fb9808c0b7 Mon Sep 17 00:00:00 2001 From: Kevin Dean <42547789+AdvancedImagingUTSW@users.noreply.github.com> Date: Mon, 7 Oct 2024 14:52:26 -0500 Subject: [PATCH 2/5] More inline type hinting... --- .../model/device_startup_functions.py | 15 ++++-- src/navigate/model/devices/daq/base.py | 5 +- src/navigate/model/devices/daq/synthetic.py | 13 ++--- src/navigate/model/microscope.py | 25 ++++++++-- src/navigate/model/model.py | 50 +++++++++++-------- src/navigate/model/plugins_model.py | 26 ++++++---- src/navigate/plugins/plugin_manager.py | 31 ++++++++---- 7 files changed, 106 insertions(+), 59 deletions(-) diff --git a/src/navigate/model/device_startup_functions.py b/src/navigate/model/device_startup_functions.py index 8b6dcd987..8fe1cc229 100644 --- a/src/navigate/model/device_startup_functions.py +++ b/src/navigate/model/device_startup_functions.py @@ -35,7 +35,7 @@ import logging import time import importlib -from multiprocessing.managers import ListProxy +from multiprocessing.managers import ListProxy, DictProxy # Third Party Imports @@ -1013,7 +1013,7 @@ def start_daq(configuration, is_synthetic=False): Parameters ---------- - configuration : multiprocesing.managers.DictProxy + configuration : multiprocessing.managers.DictProxy Global configuration of the microscope is_synthetic : bool Run synthetic version of hardware? @@ -1367,12 +1367,14 @@ def device_not_found(*args): raise RuntimeError() -def load_devices(configuration, is_synthetic=False, plugin_devices={}) -> dict: +def load_devices( + configuration: DictProxy, is_synthetic=False, plugin_devices=None +) -> dict: """Load devices from configuration. Parameters ---------- - configuration : dict + configuration : DictProxy Configuration dictionary is_synthetic : bool Run synthetic version of hardware? @@ -1382,9 +1384,12 @@ def load_devices(configuration, is_synthetic=False, plugin_devices={}) -> dict: Returns ------- devices : dict - Dictionary of devices + Dictionary of devcies """ + if plugin_devices is None: + plugin_devices = {} + devices = {} # load camera if "camera" in configuration["configuration"]["hardware"].keys(): diff --git a/src/navigate/model/devices/daq/base.py b/src/navigate/model/devices/daq/base.py index c9a6507d5..405ab59dd 100644 --- a/src/navigate/model/devices/daq/base.py +++ b/src/navigate/model/devices/daq/base.py @@ -32,6 +32,7 @@ # Standard Imports import logging +from multiprocessing.managers import DictProxy # Third Party Imports @@ -48,12 +49,12 @@ class DAQBase: """DAQBase - Parent class for Data Acquisition (DAQ) classes.""" - def __init__(self, configuration): + def __init__(self, configuration: DictProxy): """Initializes the DAQBase class. Parameters ---------- - configuration : dict + configuration : DictProxy Dictionary of configuration parameters """ diff --git a/src/navigate/model/devices/daq/synthetic.py b/src/navigate/model/devices/daq/synthetic.py index c29707fe5..b21b6d48e 100644 --- a/src/navigate/model/devices/daq/synthetic.py +++ b/src/navigate/model/devices/daq/synthetic.py @@ -36,6 +36,7 @@ from threading import Lock # Third Party Imports +from multiprocessing.managers import DictProxy # Local Imports from navigate.model.devices.daq.base import DAQBase @@ -50,12 +51,12 @@ class SyntheticDAQ(DAQBase): """SyntheticDAQ class for Data Acquisition (DAQ).""" - def __init__(self, configuration): + def __init__(self, configuration: DictProxy) -> None: """Initialize the Synthetic DAQ. Parameters ---------- - configuration : dict + configuration : DictProxy Configuration dictionary. """ super().__init__(configuration) @@ -78,7 +79,7 @@ def __init__(self, configuration): #: str: Trigger mode. Self-trigger or external-trigger. self.trigger_mode = "self-trigger" - def __str__(self): + def __str__(self) -> str: """String representation of the class.""" return "SyntheticDAQ" @@ -109,7 +110,7 @@ def close_tasks(self): """Close the tasks for triggering, analog, and counter outputs.""" pass - def prepare_acquisition(self, channel_key): + def prepare_acquisition(self, channel_key: str): """Prepare the acquisition. Parameters @@ -146,14 +147,14 @@ def write_waveforms_to_tasks(self): """Write the galvo, remote focus, and laser waveforms to each task.""" pass - def add_camera(self, microscope_name, camera): + def add_camera(self, microscope_name: str, camera: object): """Connect camera with daq: only in syntheticDAQ. Parameters ---------- microscope_name : str Name of microscope. - camera : Camera + camera : navigate.model.devices.camera.base.CameraBase Camera object. """ self.camera[microscope_name] = camera diff --git a/src/navigate/model/microscope.py b/src/navigate/model/microscope.py index 6c519b336..92ba5d732 100644 --- a/src/navigate/model/microscope.py +++ b/src/navigate/model/microscope.py @@ -32,7 +32,7 @@ # Standard Library imports import logging import importlib # noqa: F401 -from multiprocessing.managers import ListProxy +from multiprocessing.managers import ListProxy, DictProxy import reprlib # Third-party imports @@ -50,7 +50,12 @@ class Microscope: """Microscope Class - Used to control the microscope.""" def __init__( - self, name, configuration, devices_dict, is_synthetic=False, is_virtual=False + self, + name: str, + configuration: DictProxy, + devices_dict: dict, + is_synthetic=False, + is_virtual=False, ): """Initialize the microscope. @@ -58,7 +63,7 @@ def __init__( ---------- name : str Name of the microscope. - configuration : dict + configuration : DictProxy Configuration dictionary. devices_dict : dict Dictionary of devices. @@ -72,6 +77,9 @@ def __init__( #: str: Name of the microscope. self.microscope_name = name + #: Queue: Output event queue. + self.output_event_queue = None + #: dict: Configuration dictionary. self.configuration = configuration @@ -87,6 +95,9 @@ def __init__( #: bool: Ask stage for position. self.ask_stage_for_position = True + #: obj: Camera object. + self.camera = None + #: dict: Dictionary of lasers. self.lasers = {} @@ -342,7 +353,7 @@ def __init__( self.stages_list.append((stage, list(device_config["axes"]))) # connect daq and camera in synthetic mode - if is_synthetic: + if is_synthetic and self.daq is not None: self.daq.add_camera(self.microscope_name, self.camera) def update_data_buffer(self, data_buffer, number_of_frames): @@ -850,7 +861,9 @@ def update_stage_limits(self, limits_flag=True) -> None: for stage, _ in self.stages_list: stage.stage_limits = limits_flag - def assemble_device_config_lists(self, device_name: str, device_name_dict: dict): + def assemble_device_config_lists( + self, device_name: str, device_name_dict: dict + ) -> tuple: """Assemble device config lists. Parameters @@ -866,6 +879,8 @@ def assemble_device_config_lists(self, device_name: str, device_name_dict: dict) Device configuration list. device_name_list : list Device name list. + is_list : bool + Is list. """ device_config_list = [] device_name_list = [] diff --git a/src/navigate/model/model.py b/src/navigate/model/model.py index 8b1f47f90..c3764eff9 100644 --- a/src/navigate/model/model.py +++ b/src/navigate/model/model.py @@ -1,5 +1,6 @@ # Copyright (c) 2021-2024 The University of Texas Southwestern Medical Center. # All rights reserved. +import argparse # Redistribution and use in source and binary forms, with or without # modification, are permitted for academic and research use only (subject to the @@ -88,38 +89,40 @@ class Model: Model for Model-View-Controller Software Architecture.""" - def __init__(self, args, configuration=None, event_queue=None): + def __init__(self, args: argparse.Namespace, configuration=None, event_queue=None): """Initialize the Model. Parameters ---------- args : argparse.Namespace Command line arguments. - configuration : dict + configuration : multiprocessing.managers.DictProxy Configuration dictionary. event_queue : multiprocessing.Queue - Queue for events. + Event queue. Receives events from the controller. """ - + # Set up logging log_setup("model_logging.yml") + #: object: Logger object. self.logger = logging.getLogger(p) - # Loads the YAML file for all of the microscope parameters #: dict: Configuration dictionary. self.configuration = configuration + # Plugins plugins = PluginsModel() - # load plugin feature and devices plugin_devices, plugin_acquisition_modes = plugins.load_plugins() + + #: dict: Dictionary of plugin acquisition modes + self.plugin_acquisition_modes = plugin_acquisition_modes + + # Devices devices_dict = load_devices( configuration, args.synthetic_hardware, plugin_devices ) devices_dict["__plugins__"] = plugin_devices - #: dict: Dictionary of plugin acquisition modes - self.plugin_acquisition_modes = plugin_acquisition_modes - #: dict: Dictionary of virtual microscopes. self.virtual_microscopes = {} @@ -744,9 +747,7 @@ def run_command(self, command, *args, **kwargs): self, self.addon_feature ) except KeyError: - self.logger.debug( - f"Attempted to load an unknown feature: {args}." - ) + self.logger.debug(f"Attempted to load an unknown feature: {args}.") elif command == "stage_limits": for microscope_name in self.microscopes: self.microscopes[microscope_name].update_stage_limits(args[0]) @@ -825,8 +826,9 @@ def move_stage(self, pos_dict, wait_until_done=False): """ try: r = self.active_microscope.move_stage(pos_dict, wait_until_done) - self.logger.info(f"Stage moved to:, {pos_dict}, " - f"Wait until done: {wait_until_done}") + self.logger.info( + f"Stage moved to:, {pos_dict}, " f"Wait until done: {wait_until_done}" + ) except Exception as e: self.logger.debug(f"Stage move failed: {e}") return False @@ -905,13 +907,17 @@ def run_data_process(self, num_of_frames=0, data_func=None): self.logger.info(f"Running data process, getting frames {frame_ids}") # if there is at least one frame available if not frame_ids: - self.logger.debug(f"Frame not received. Waiting {wait_num}" - f"/{self.camera_wait_iterations} iterations") + self.logger.debug( + f"Frame not received. Waiting {wait_num}" + f"/{self.camera_wait_iterations} iterations" + ) wait_num -= 1 if wait_num <= 0: - error_statement = ("Acquisition aborted due to camera time out " - "error. Please verify that the external " - "trigger is connected and configured properly.") + error_statement = ( + "Acquisition aborted due to camera time out " + "error. Please verify that the external " + "trigger is connected and configured properly." + ) self.logger.debug(error_statement) print(error_statement) @@ -1216,8 +1222,10 @@ def change_resolution(self, resolution_value): self.stop_stage() except ValueError as e: - self.logger.debug(f"Error changing microscope resolution:" - f".{self.active_microscope_name} - {e}") + self.logger.debug( + f"Error changing microscope resolution:" + f".{self.active_microscope_name} - {e}" + ) self.active_microscope.ask_stage_for_position = True diff --git a/src/navigate/model/plugins_model.py b/src/navigate/model/plugins_model.py index d57f91e0a..4c3a7f458 100644 --- a/src/navigate/model/plugins_model.py +++ b/src/navigate/model/plugins_model.py @@ -33,6 +33,7 @@ # Standard library imports import os from pathlib import Path +from typing import Union # Third-party imports @@ -46,15 +47,19 @@ class PluginsModel: """Plugins manager in the model side""" - def __init__(self): + def __init__(self) -> None: """Initialize plugins manager class""" + #: dict: devices dictionary self.devices_dict = {} - self.plugin_acquisition_modes = {} - def load_plugins(self): - """Load plugins""" + #: dict: plugin acquisition modes dictionary + self.plugin_acquisition_modes = {} + #: str: feature lists path self.feature_lists_path = os.path.join(get_navigate_path(), "feature_lists") + + def load_plugins(self) -> tuple: + """Load plugins""" if not os.path.exists(self.feature_lists_path): os.makedirs(self.feature_lists_path) @@ -65,16 +70,19 @@ def load_plugins(self): plugin_file_manager = PluginFileManager(plugins_path, plugins_config_path) self.load_plugins_through_manager(plugin_file_manager) self.load_plugins_through_manager(PluginPackageManager) - return self.devices_dict, self.plugin_acquisition_modes - def load_plugins_through_manager(self, plugin_manager): + def load_plugins_through_manager( + self, plugin_manager: Union[PluginFileManager, PluginPackageManager] + ) -> None: """Load plugins through plugin manager Parameters - ---------- - plugin_manager : object - PluginManager object + ------- + plugin_manager : PluginFileManager or PluginPackageManager + - PluginFileManager + - PluginPackageManager + """ plugins = plugin_manager.get_plugins() diff --git a/src/navigate/plugins/plugin_manager.py b/src/navigate/plugins/plugin_manager.py index 3157debcd..d115cf162 100644 --- a/src/navigate/plugins/plugin_manager.py +++ b/src/navigate/plugins/plugin_manager.py @@ -38,6 +38,7 @@ import os import inspect import logging +from typing import Optional # Local application imports from navigate.tools.file_functions import load_yaml_file @@ -49,7 +50,7 @@ logger = logging.getLogger(p) -def register_features(module): +def register_features(module) -> None: """Register features Parameters @@ -66,7 +67,7 @@ class PluginPackageManager: """Plugin package manager""" @staticmethod - def get_plugins(): + def get_plugins() -> dict: """Get plugins Returns @@ -89,7 +90,7 @@ def get_plugins(): return plugins @staticmethod - def load_config(package_name): + def load_config(package_name: str) -> dict: """Load plugin_config.yml Parameters @@ -107,7 +108,7 @@ def load_config(package_name): return plugin_config @staticmethod - def load_controller(package_name, controller_name): + def load_controller(package_name: str, controller_name: str) -> Optional[type]: """Load controller Parameters @@ -119,8 +120,12 @@ def load_controller(package_name, controller_name): Returns ------- - controller_class : class - controller class + controller_class : type or None + The dynamically loaded controller class, or `None` if the controller + module or class cannot be found. The return type can be: + + - `type` : If the controller class is found, the class type is returned. + - `None` : If the module or class cannot be found, `None` is returned. """ controller_file_name = "_".join(controller_name.lower().split()) + "_controller" controller_class_name = "".join(controller_name.title().split()) + "Controller" @@ -134,7 +139,7 @@ def load_controller(package_name, controller_name): return None @staticmethod - def load_view(package_name, frame_name): + def load_view(package_name: str, frame_name: str) -> Optional[type]: """Load view Parameters @@ -146,8 +151,12 @@ def load_view(package_name, frame_name): Returns ------- - frame_class : class - frame class + view_class : type or None + The dynamically loaded view class, or `None` if the view module or + class cannot be found. The return type can be: + + - `type` : If the view class is found, the class type is returned. + - `None` : If the module or class cannot be found, `None` is returned. """ frame_file_name = "_".join(frame_name.lower().split()) + "_frame" frame_class_name = "".join(frame_name.title().split()) + "Frame" @@ -161,7 +170,7 @@ def load_view(package_name, frame_name): return None @staticmethod - def load_feature_lists(package_name, register_func): + def load_feature_lists(package_name, register_func) -> None: """Load feature lists Parameters @@ -255,7 +264,7 @@ def load_devices(package_name, register_func): class PluginFileManager: """Plugin file manager""" - def __init__(self, plugins_path, plugins_config_path): + def __init__(self, plugins_path: str, plugins_config_path: str) -> None: """Initialize PluginFileManager Parameters From 5ac6bf62697f9fd025f532867163685c001ee18b Mon Sep 17 00:00:00 2001 From: Kevin Dean <42547789+AdvancedImagingUTSW@users.noreply.github.com> Date: Tue, 8 Oct 2024 06:01:52 -0500 Subject: [PATCH 3/5] More of the same... Interesting. Can specify a base class as the return object to provide access to methods and documentation... --- .../model/device_startup_functions.py | 508 +++++++++++------- src/navigate/model/devices/daq/ni.py | 5 +- src/navigate/model/devices/galvo/base.py | 18 +- src/navigate/model/devices/galvo/ni.py | 18 +- src/navigate/model/devices/galvo/synthetic.py | 18 +- src/navigate/model/devices/lasers/base.py | 17 +- src/navigate/model/devices/lasers/ni.py | 10 +- .../model/devices/lasers/synthetic.py | 16 +- .../model/devices/remote_focus/base.py | 12 +- .../remote_focus/equipment_solutions.py | 12 +- src/navigate/model/devices/remote_focus/ni.py | 10 +- .../model/devices/remote_focus/synthetic.py | 13 +- src/navigate/model/devices/stages/base.py | 21 +- src/navigate/model/devices/stages/sutter.py | 22 +- src/navigate/tools/file_functions.py | 41 +- 15 files changed, 457 insertions(+), 284 deletions(-) diff --git a/src/navigate/model/device_startup_functions.py b/src/navigate/model/device_startup_functions.py index 8fe1cc229..c13d2ea3e 100644 --- a/src/navigate/model/device_startup_functions.py +++ b/src/navigate/model/device_startup_functions.py @@ -29,18 +29,28 @@ # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE # POSSIBILITY OF SUCH DAMAGE. - # Standard Library Imports import platform import logging import time import importlib from multiprocessing.managers import ListProxy, DictProxy +from typing import Callable, Tuple, Any, Type # Third Party Imports # Local Imports from navigate.tools.common_functions import build_ref_name +from navigate.model.devices.camera.base import CameraBase +from navigate.model.devices.daq.base import DAQBase +from navigate.model.devices.filter_wheel.base import FilterWheelBase +from navigate.model.devices.galvo.base import GalvoBase +from navigate.model.devices.lasers.base import LaserBase +from navigate.model.devices.mirrors.base import MirrorBase +from navigate.model.devices.shutter.base import ShutterBase +from navigate.model.devices.stages.base import StageBase +from navigate.model.devices.remote_focus.base import RemoteFocusBase +from navigate.model.devices.zoom.base import ZoomBase # Logger Setup p = __name__.split(".")[1] @@ -53,24 +63,44 @@ class DummyDeviceConnection: pass -def auto_redial(func, args, n_tries=10, exception=Exception, **kwargs): - """Retries connections to a startup device defined by func n_tries times. +def auto_redial( + func: Callable[..., Any], + args: Tuple[Any, ...], + n_tries: int = 10, + exception: Type[Exception] = Exception, + **kwargs: Any, +) -> Any: + """Retries connections to a startup device defined by `func` for a specified + number of attempts. + + This function attempts to execute the connection function `func` up to `n_tries` + times. If an exception occurs, it retries the connection after a brief pause, + logging each failure. If the connection partially succeeds, it cleans up any objects + before retrying. Parameters ---------- - func : function or class - The function or class (__init__() function) that connects to a device. - args : tuple - Arguments to function or class - n_tries : int - The number of tries to redial. - exception : inherits from BaseException - An exception type to check on each connection attempt. + func : Callable[..., Any] + The function or class (`__init__()` method) that connects to a device. + args : Tuple[Any, ...] + Positional arguments to pass to the `func`. + n_tries : int, Optional + The number of attempts to retry the connection. Default is 10. + exception : Type[Exception], Optional + The exception type to catch and handle during connection attempts. + Default is `Exception`. + **kwargs : Any + Additional keyword arguments passed to `func`. Returns ------- - val : object - Result of func + Any + The result of the successful execution of `func`. + + Raises + ------ + exception + If all connection attempts fail, the specified `exception` is raised. """ val = None @@ -109,22 +139,38 @@ class SerialConnectionFactory: _connections = {} @classmethod - def build_connection(cls, build_connection_function, args, exception=Exception): - """Builds a serial connection to a device. + def build_connection( + cls, + build_connection_function: Callable[..., Any], + args: Tuple[Any, ...], + exception: Type[Exception] = Exception, + ) -> Any: + """ + Builds a serial connection to a device. + + This method establishes a connection to a device using the provided + connection-building function and arguments. If the connection does not exist, + it will be created and stored. Parameters ---------- - build_connection_function : function + build_connection_function : Callable Function that builds the connection to the device. - args : tuple + args : Tuple Arguments to the build_connection_function - exception : Exception + exception : Exception, Optional Exception to catch when building the connection Returns ------- - connection : object + connection : Any Connection to the device + + Raises + ------ + exception : Exception + If the connection building process fails, the specified `exception` is + raised. """ port = args[0] if str(port) not in cls._connections: @@ -135,25 +181,27 @@ def build_connection(cls, build_connection_function, args, exception=Exception): return cls._connections[str(port)] -def load_camera_connection(configuration, camera_id=0, is_synthetic=False): - """Initializes the camera api class. +def load_camera_connection( + configuration: DictProxy, camera_id: int = 0, is_synthetic: bool = False +) -> Any: + """Initialize the camera API class. Load camera information from the configuration file. Proper camera types include HamamatsuOrca, HamamatsuOrcaLightning, Photometrics, and SyntheticCamera. Parameters ---------- - configuration : multiprocessing.managers.DictProxy + configuration : DictProxy Global configuration of the microscope - camera_id : int + camera_id : int, Optional Device ID (0, 1...) - is_synthetic: bool + is_synthetic: bool, Optional Whether it is a synthetic hardware Returns ------- - Camera controller: class - Camera api class. + Camera_controller: Any + The initialized camera API class instance. """ if is_synthetic: @@ -198,32 +246,35 @@ def load_camera_connection(configuration, camera_id=0, is_synthetic=False): def start_camera( - microscope_name, - device_connection, - configuration, - is_synthetic=False, - plugin_devices={}, -): - """Initializes the camera class. + microscope_name: str, + device_connection: Any, + configuration: DictProxy, + is_synthetic: bool = False, + plugin_devices=None, +) -> CameraBase: + """Initialize the camera class. Parameters ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocessing.managers.DictProxy + configuration : DictProxy Global configuration of the microscope - is_synthetic : bool - Run synthetic version of hardware? - plugin_devices : dict - Dictionary of plugin devices + is_synthetic : bool, Optional + Run synthetic version of hardware. Default is False. + plugin_devices : dict, Optional + Dictionary of plugin devices. Default is None. Returns ------- - Camera : class - Camera class. + Camera : CameraBase + Instantiated camera class. """ + if plugin_devices is None: + plugin_devices = {} + if device_connection is None: device_not_found(microscope_name, "camera") @@ -281,27 +332,25 @@ def start_camera( device_not_found(microscope_name, "camera", cam_type) -def load_mirror(configuration, is_synthetic=False): - """Initializes the deformable mirror class on a dedicated thread. +def load_mirror(configuration: DictProxy, is_synthetic: bool = False) -> Any: + """Initializes the deformable mirror API. Parameters ---------- - configuration : multiprocesing.managers.DictProxy + configuration : DictProxy Global configuration of the microscope - is_synthetic : bool - Run synthetic version of hardware? + is_synthetic : bool, Optional + Run synthetic version of hardware. Default is False. Returns ------- - Mirror : class - Mirror class. + mirror : Any + Instantiated . """ if is_synthetic: mirror_type = "SyntheticMirror" else: - mirror_type = configuration["configuration"]["hardware"]["mirror"][ - "type" - ] # only one mirror for now... + mirror_type = configuration["configuration"]["hardware"]["mirror"]["type"] if mirror_type == "ImagineOpticsMirror": from navigate.model.devices.APIs.imagineoptics.imop import IMOP_Mirror @@ -316,32 +365,30 @@ def load_mirror(configuration, is_synthetic=False): def start_mirror( - microscope_name, - device_connection, - configuration, - is_synthetic=False, + microscope_name: str, + device_connection: Any, + configuration: DictProxy, + is_synthetic: bool = False, plugin_devices={}, -): - """Initializes the mirror class. +) -> MirrorBase: + """Initialize the mirror class. - Parameters - ---------- Parameters ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocesing.managers.DictProxy + configuration : DictProxy Global configuration of the microscope - is_synthetic : bool - Run synthetic version of hardware? + is_synthetic : bool, Optional + Run synthetic version of hardware. Default is False. plugin_devices : dict Dictionary of plugin devices Returns ------- - Mirror : class + mirror : Any Mirror class. """ if device_connection is None or is_synthetic: @@ -371,26 +418,31 @@ def start_mirror( device_not_found(microscope_name, "mirror", mirror_type) -def load_stages(configuration, is_synthetic=False, plugin_devices={}): - """Initializes the stage class on a dedicated thread. +def load_stages( + configuration: DictProxy, is_synthetic: bool = False, plugin_devices=None +) -> Any: + """Initialize the stage API. Stage information is pulled from the configuration file. Proper stage types include PI, MP285, Thorlabs, MCL, ASI, GalvoNIStage, and SyntheticStage. Parameters ---------- - configuration : multiprocessing.managers.DictProxy + configuration : DictProxy Global configuration of the microscope - is_synthetic : bool - Run synthetic version of hardware? - plugin_devices : dict - Dictionary of plugin devices + is_synthetic : bool, Optional + Run synthetic version of hardware. Default is False. + plugin_devices : dict, Optional + Dictionary of plugin devices. Default is None. Returns ------- - Stage : class + Stage : Any Stage class. """ + if plugin_devices is None: + plugin_devices = {} + stage_devices = [] stages = configuration["configuration"]["hardware"]["stage"] @@ -587,36 +639,42 @@ def load_stages(configuration, is_synthetic=False, plugin_devices={}): def start_stage( - microscope_name, - device_connection, - configuration, - id=0, - is_synthetic=False, - plugin_devices={}, -): - """Initializes the Stage class. Proper stage types include PI, MP285, Thorlabs, MCL, - ASI, GalvoNIStage, and SyntheticStage. + microscope_name: str, + device_connection: Any, + configuration: DictProxy, + id: int = 0, + is_synthetic: bool = False, + plugin_devices=None, +) -> StageBase: + """Initialize the Stage class. + + Proper stage types include PI, MP285, Thorlabs, MCL, ASI, GalvoNIStage, + and SyntheticStage. Parameters ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocesing.managers.DictProxy + configuration : DictProxy Global configuration of the microscope - id : int - ID of the stage - is_synthetic : bool - Run synthetic version of hardware? - plugin_devices : dict - Dictionary of plugin devices + id : int, Optional + ID of the stage. Default is 0. + is_synthetic : bool, Optional + Run synthetic version of hardware. Default is False. + plugin_devices : dict, Optional + Dictionary of plugin devices. Default is None. Returns ------- - Stage : class - Stage class. + Stage : StageBase + An instance of the appropriate stage class depending on the device + configuration. """ + if plugin_devices is None: + plugin_devices = {} + device_config = configuration["configuration"]["microscopes"][microscope_name][ "stage" ]["hardware"] @@ -698,7 +756,9 @@ def start_stage( device_not_found(microscope_name, "stage", device_type, id) -def load_zoom_connection(configuration, is_synthetic=False, plugin_devices={}): +def load_zoom_connection( + configuration: DictProxy, is_synthetic: bool = False, plugin_devices=None +) -> Any: """Initializes the Zoom class on a dedicated thread. Load zoom information from the configuration file. Proper zoom types include @@ -706,12 +766,12 @@ def load_zoom_connection(configuration, is_synthetic=False, plugin_devices={}): Parameters ---------- - configuration : multiprocesing.managers.DictProxy + configuration : DictProxy Global configuration of the microscope is_synthetic : bool Run synthetic version of hardware? - plugin_devices : dict - Dictionary of plugin devices + plugin_devices : dict, Optional + Dictionary of plugin devices. Default is None. Returns ------- @@ -719,6 +779,9 @@ def load_zoom_connection(configuration, is_synthetic=False, plugin_devices={}): Zoom class. """ + if plugin_devices is None: + plugin_devices = {} + device_info = configuration["configuration"]["hardware"]["zoom"] if is_synthetic: device_type = "SyntheticZoom" @@ -751,12 +814,12 @@ def load_zoom_connection(configuration, is_synthetic=False, plugin_devices={}): def start_zoom( - microscope_name, - device_connection, - configuration, - is_synthetic=False, - plugin_devices={}, -): + microscope_name: str, + device_connection: Any, + configuration: DictProxy, + is_synthetic: bool = False, + plugin_devices=None, +) -> ZoomBase: """Initializes the zoom class on a dedicated thread. Initializes the zoom: DynamixelZoom and SyntheticZoom. @@ -765,20 +828,23 @@ def start_zoom( ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocesing.managers.DictProxy + configuration : DictProxy Global configuration of the microscope - is_synthetic : bool - Run synthetic version of hardware? - plugin_devices : dict - Dictionary of plugin devices + is_synthetic : bool, Optional + Run synthetic version of hardware. Default is False. + plugin_devices : dict, Optional + Dictionary of plugin devices. Default is None. Returns ------- - Zoom : class + Zoom : ZoomBase Zoom class. """ + if plugin_devices is None: + plugin_devices = {} + if is_synthetic: device_type = "SyntheticZoom" @@ -826,7 +892,9 @@ def start_zoom( device_not_found("Zoom", device_type) -def load_filter_wheel_connection(device_info, is_synthetic=False, plugin_devices={}): +def load_filter_wheel_connection( + device_info: DictProxy, is_synthetic=False, plugin_devices=None +) -> Any: """Initializes the Filter Wheel class on a dedicated thread. Load filter wheel information from the configuration file. Proper filter wheel types @@ -834,18 +902,21 @@ def load_filter_wheel_connection(device_info, is_synthetic=False, plugin_devices Parameters ---------- - device_info : multiprocesing.managers.DictProxy + device_info : DictProxy filter wheel device configuration - is_synthetic : bool - Run synthetic version of hardware? - plugin_devices : dict - Dictionary of plugin devices + is_synthetic : bool, Optional + Run synthetic version of hardware, Default is False. + plugin_devices : dict, Optional + Dictionary of plugin devices. Default is None. Returns ------- - Filter : class + Filter : Any Filter class. """ + if plugin_devices is None: + plugin_devices = {} + if is_synthetic: device_type = "SyntheticFilterWheel" @@ -909,14 +980,14 @@ def load_filter_wheel_connection(device_info, is_synthetic=False, plugin_devices def start_filter_wheel( - microscope_name, - device_connection, - configuration, - id=0, - is_synthetic=False, - plugin_devices={}, -): - """Initializes the filter wheel class on a dedicated thread. + microscope_name: str, + device_connection: Any, + configuration: DictProxy, + id: int = 0, + is_synthetic: bool = False, + plugin_devices=None, +) -> FilterWheelBase: + """Initialize the filter wheel class. Initializes the filter wheel: SutterFilterWheel, ASI, and SyntheticFilterWheel. @@ -924,22 +995,25 @@ def start_filter_wheel( ---------- microscope_name : str Name of microscope in configuration - device_connection : object - Hardware device to connect to - configuration : multiprocesing.managers.DictProxy + device_connection : Any + Device connection + configuration : DictProxy Global configuration of the microscope - id : int - Index of filter_wheel in the configuration dictionary - is_synthetic : bool - Run synthetic version of hardware? - plugin_devices : dict - Dictionary of plugin devices + id : int, Optional + Index of filter_wheel in the configuration dictionary. Default is 0. + is_synthetic : bool, Optional + Run synthetic version of hardware. Default is False. + plugin_devices : dict, Optional + Dictionary of plugin devices. Default is None. Returns ------- - FilterWheel : class + FilterWheel : FilterWheelBase FilterWheel class. """ + if plugin_devices is None: + plugin_devices = {} + if device_connection is None: device_not_found(microscope_name, "filter_wheel") @@ -1005,7 +1079,7 @@ def start_filter_wheel( device_not_found(microscope_name, "filter_wheel", device_type) -def start_daq(configuration, is_synthetic=False): +def start_daq(configuration: DictProxy, is_synthetic: bool = False) -> DAQBase: """Initializes the data acquisition (DAQ) class on a dedicated thread. Load daq information from the configuration file. Proper daq types include NI and @@ -1013,14 +1087,14 @@ def start_daq(configuration, is_synthetic=False): Parameters ---------- - configuration : multiprocessing.managers.DictProxy + configuration : DictProxy Global configuration of the microscope - is_synthetic : bool - Run synthetic version of hardware? + is_synthetic : bool, Optional + Run synthetic version of hardware. Default is False. Returns ------- - DAQ : class + DAQ : DAQBase DAQ class. """ if is_synthetic: @@ -1044,12 +1118,12 @@ def start_daq(configuration, is_synthetic=False): def start_shutter( - microscope_name, - device_connection, - configuration, - is_synthetic=False, - plugin_devices={}, -): + microscope_name: str, + device_connection: Any, + configuration: DictProxy, + is_synthetic: bool = False, + plugin_devices: dict = None, +) -> ShutterBase: """Initializes the shutter class on a dedicated thread. Initializes the shutters: ThorlabsShutter or SyntheticShutter @@ -1060,21 +1134,23 @@ def start_shutter( ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocessing.managers.DictProxy + configuration : DictProxy Global configuration of the microscope - is_synthetic : bool - Run synthetic version of hardware? - plugin_devices : dict - Dictionary of plugin devices + is_synthetic : bool, Optional + Run synthetic version of hardware. Default is False. + plugin_devices : dict, Optional + Dictionary of plugin devices. Default is None. Returns ------- - Shutter : class + Shutter : ShutterBase Shutter class. """ + if plugin_devices is None: + plugin_devices = {} if is_synthetic: device_type = "SyntheticShutter" @@ -1118,13 +1194,13 @@ def start_shutter( def start_lasers( - microscope_name, - device_connection, - configuration, - id=0, - is_synthetic=False, - plugin_devices={}, -): + microscope_name: str, + device_connection: Any, + configuration: DictProxy, + id: int = 0, + is_synthetic: bool = False, + plugin_devices: dict = None, +) -> LaserBase: """Initializes the lasers. Loads the lasers from the configuration file. Proper laser types include NI and @@ -1134,23 +1210,26 @@ def start_lasers( ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocesing.managers.DictProxy + configuration : DictProxy Global configuration of the microscope - id : int - Index of laser in laser list in configuration dictionary - is_synthetic : bool - Run synthetic version of hardware? - plugin_devices : dict - Dictionary of plugin devices + id : int, Optional + Index of laser in laser list in configuration dictionary. Default is 0. + is_synthetic : bool, Optional + Run synthetic version of hardware. Default is False. + plugin_devices : dict, Optional + Dictionary of plugin devices. Default is None. Returns ------- - Triggers : class - Trigger class. + laser : LaserBase + Laser class. """ + if plugin_devices is None: + plugin_devices = {} + if is_synthetic: device_type = "SyntheticLaser" @@ -1193,12 +1272,12 @@ def start_lasers( def start_remote_focus_device( - microscope_name, - device_connection, - configuration, - is_synthetic=False, - plugin_devices={}, -): + microscope_name: str, + device_connection: Any, + configuration: DictProxy, + is_synthetic: bool = False, + plugin_devices=None, +) -> RemoteFocusBase: """Initializes the remote focus class. Initializes the Remote Focusing Device. Proper remote focus types include @@ -1208,21 +1287,24 @@ def start_remote_focus_device( ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocesing.managers.DictProxy + configuration : DictProxy Global configuration of the microscope - is_synthetic : bool - Run synthetic version of hardware? - plugin_devices : dict - Dictionary of plugin devices + is_synthetic : bool, Optional + Run synthetic version of hardware. Default is False. + plugin_devices : dict, Optional + Dictionary of plugin devices. Default is None. Returns ------- - Remote Focus : class + remote_focus : RemoteFocusBase Remote focusing class. """ + if plugin_devices is None: + plugin_devices = {} + if is_synthetic: device_type = "SyntheticRemoteFocus" @@ -1274,13 +1356,13 @@ def start_remote_focus_device( def start_galvo( - microscope_name, - device_connection, - configuration, - id=0, - is_synthetic=False, - plugin_devices={}, -): + microscope_name: str, + device_connection: Any, + configuration: DictProxy, + id: int = 0, + is_synthetic: bool = False, + plugin_devices=None, +) -> GalvoBase: """Initializes the Galvo class. Initializes the Galvo Device. Proper galvo types include NI and SyntheticGalvo. @@ -1289,23 +1371,26 @@ def start_galvo( ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocessing.managers.DictProxy + configuration : DictProxy Global configuration of the microscope - id : int - Index of galvo in the configuration dictionary - is_synthetic : bool - Run synthetic version of hardware? - plugin_devices : dict - Dictionary of plugin devices + id : int, Optional + Index of galvo in the configuration dictionary. Default is 0. + is_synthetic : bool, Optional + Run synthetic version of hardware. Default is False. + plugin_devices : dict, Optional + Dictionary of plugin devices. Default is None. Returns ------- - Galvo : class + Galvo : GalvoBase Galvo scanning class. """ + if plugin_devices is None: + plugin_devices = {} + if is_synthetic: device_type = "SyntheticGalvo" @@ -1343,23 +1428,28 @@ def start_galvo( device_not_found(microscope_name, "galvo", id, device_type) -def device_not_found(*args): - """Display error message if device not found. +def device_not_found(*args: Any) -> None: + """ + Display an error message and raise an exception if the device is not found. - Raises error if the configuration is incorrect and/or the device is not found. + This function logs an error message and raises a `RuntimeError` when the + specified device cannot be found in the configuration or the configuration is + incorrect. Parameters ---------- - *args - microscope_name - device - device id - device type + args : tuple + A list of arguments representing the device details. These are typically: - Returns - ------- - devices : class - Device class. + - microscope_name (str): The name of the microscope. + - device (str): The name of the device. + - device id (int): The ID of the device. + - device type (str): The type of the device. + + Raises + ------ + RuntimeError + If the device cannot be found or the configuration is incorrect. """ error_statement = f"Device not found in configuration: {args}" logger.error(error_statement) diff --git a/src/navigate/model/devices/daq/ni.py b/src/navigate/model/devices/daq/ni.py index 2ad48550f..b9c490471 100644 --- a/src/navigate/model/devices/daq/ni.py +++ b/src/navigate/model/devices/daq/ni.py @@ -32,6 +32,7 @@ # Standard Imports import logging +import multiprocessing.managers from threading import Lock import traceback import time @@ -57,12 +58,12 @@ class NIDAQ(DAQBase): """NIDAQ class for Control of NI Data Acquisition Cards.""" - def __init__(self, configuration: dict) -> None: + def __init__(self, configuration: multiprocessing.managers.DictProxy) -> None: """Initialize NIDAQ class. Parameters ---------- - configuration : dict + configuration : multiprocessing.managers.DictProxy Configuration dictionary. """ super().__init__(configuration) diff --git a/src/navigate/model/devices/galvo/base.py b/src/navigate/model/devices/galvo/base.py index ade53319d..0544940aa 100644 --- a/src/navigate/model/devices/galvo/base.py +++ b/src/navigate/model/devices/galvo/base.py @@ -32,6 +32,8 @@ # Standard Library Imports import logging +from typing import Any +from multiprocessing.managers import DictProxy # Third Party Imports @@ -48,19 +50,25 @@ class GalvoBase: """GalvoBase Class - Parent class for galvanometers.""" - def __init__(self, microscope_name, device_connection, configuration, galvo_id=0): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: DictProxy, + galvo_id: int = 0, + ) -> None: """Initialize the GalvoBase class. Parameters ---------- microscope_name : str Name of the microscope. - device_connection : dict - Dictionary of device connections. - configuration : dict + device_connection : Any + Device connection. + configuration : DictProxy Dictionary of configuration parameters. galvo_id : int - Galvo ID. + Galvo ID. Default is 0. """ #: dict: Dictionary of microscope configuration parameters. diff --git a/src/navigate/model/devices/galvo/ni.py b/src/navigate/model/devices/galvo/ni.py index 5d04ec3bb..386d3769d 100644 --- a/src/navigate/model/devices/galvo/ni.py +++ b/src/navigate/model/devices/galvo/ni.py @@ -32,6 +32,8 @@ # Standard Library Imports import logging +from typing import Any +from multiprocessing.managers import DictProxy # Third Party Imports import nidaqmx @@ -49,19 +51,25 @@ class GalvoNI(GalvoBase): """GalvoNI Class - NI DAQ Control of Galvanometers""" - def __init__(self, microscope_name, device_connection, configuration, galvo_id=0): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: DictProxy, + galvo_id: int = 0, + ) -> None: """Initialize the GalvoNI class. Parameters ---------- microscope_name : str Name of the microscope. - device_connection : dict - Dictionary of device connections. - configuration : dict + device_connection : Any + Connection to the NI DAQ device. + configuration : DictProxy Dictionary of configuration parameters. galvo_id : int - Galvo ID. + Galvo ID. Default is 0. """ super().__init__(microscope_name, device_connection, configuration, galvo_id) diff --git a/src/navigate/model/devices/galvo/synthetic.py b/src/navigate/model/devices/galvo/synthetic.py index aa1aa38d4..8065d6941 100644 --- a/src/navigate/model/devices/galvo/synthetic.py +++ b/src/navigate/model/devices/galvo/synthetic.py @@ -32,6 +32,8 @@ # Standard Library Imports import logging +from typing import Any +from multiprocessing.managers import DictProxy # Third Party Imports @@ -48,19 +50,25 @@ class SyntheticGalvo(GalvoBase): """SyntheticGalvo Class""" - def __init__(self, microscope_name, device_connection, configuration, galvo_id=0): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: DictProxy, + galvo_id: int = 0, + ) -> None: """Initialize the SyntheticGalvo class. Parameters ---------- microscope_name : str Name of the microscope. - device_connection : dict - Dictionary of device connections. - configuration : dict + device_connection : Any + Device connection. + configuration : DictProxy Dictionary of configuration parameters. galvo_id : int - Galvo ID. + Galvo ID. Default is 0. """ super().__init__(microscope_name, device_connection, configuration, galvo_id) diff --git a/src/navigate/model/devices/lasers/base.py b/src/navigate/model/devices/lasers/base.py index 30f6b4791..02da3be5d 100644 --- a/src/navigate/model/devices/lasers/base.py +++ b/src/navigate/model/devices/lasers/base.py @@ -30,11 +30,10 @@ # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE # POSSIBILITY OF SUCH DAMAGE. -""" -Laser Base Class -""" # Standard Library Imports import logging +from typing import Any +from multiprocessing.managers import DictProxy # Third Party Imports @@ -50,16 +49,22 @@ class LaserBase: """Laser Base Class""" - def __init__(self, microscope_name, device_connection, configuration, laser_id): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: DictProxy, + laser_id: int, + ) -> None: """Initialize Laser Base Class Parameters ---------- microscope_name : str Name of the microscope - device_connection : object + device_connection : Any Communication instance with the device. - configuration : dict + configuration : DictProxy Configuration dictionary laser_id : int Laser ID diff --git a/src/navigate/model/devices/lasers/ni.py b/src/navigate/model/devices/lasers/ni.py index a69e4d272..d6512d2c4 100644 --- a/src/navigate/model/devices/lasers/ni.py +++ b/src/navigate/model/devices/lasers/ni.py @@ -32,6 +32,8 @@ # Standard Library Imports import logging +from typing import Any +from multiprocessing.managers import DictProxy # Third Party Imports import nidaqmx @@ -57,8 +59,8 @@ class LaserNI(LaserBase): def __init__( self, microscope_name: str, - device_connection: object, - configuration: dict, + device_connection: Any, + configuration: DictProxy, laser_id: int, ) -> None: """Initialize the LaserNI class. @@ -67,9 +69,9 @@ def __init__( ---------- microscope_name : str The microscope name. - device_connection : object + device_connection : Any The device connection object. - configuration : dict + configuration : DictProxy The device configuration. laser_id : int The laser id. diff --git a/src/navigate/model/devices/lasers/synthetic.py b/src/navigate/model/devices/lasers/synthetic.py index c6e24056b..c89fc4e99 100644 --- a/src/navigate/model/devices/lasers/synthetic.py +++ b/src/navigate/model/devices/lasers/synthetic.py @@ -32,6 +32,8 @@ # Standard Library Imports import logging +from typing import Any +from multiprocessing.managers import DictProxy # Third Party Imports @@ -48,18 +50,24 @@ class SyntheticLaser(LaserBase): """SyntheticLaser Class""" - def __init__(self, microscope_name, device_connection, configuration, laser_id): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: DictProxy, + laser_id: int, + ) -> None: """Initialize the SyntheticLaser class. Parameters ---------- microscope_name : str The microscope name. - device_connection : object + device_connection : Any The device connection object. - configuration : dict + configuration : DictProxy The device configuration. - laser_id : str + laser_id : int The laser ID. """ super().__init__(microscope_name, device_connection, configuration, laser_id) diff --git a/src/navigate/model/devices/remote_focus/base.py b/src/navigate/model/devices/remote_focus/base.py index 0586505bd..323165b66 100644 --- a/src/navigate/model/devices/remote_focus/base.py +++ b/src/navigate/model/devices/remote_focus/base.py @@ -32,6 +32,8 @@ # Standard Library Imports import logging +from typing import Any +from multiprocessing.managers import DictProxy # Third Party Imports @@ -52,16 +54,18 @@ class RemoteFocusBase: """RemoteFocusBase Class - Parent class for Remote Focusing Device.""" - def __init__(self, microscope_name, device_connection, configuration): + def __init__( + self, microscope_name: str, device_connection: Any, configuration: DictProxy + ) -> None: """Initializes the RemoteFocusBase Class. Parameters ---------- microscope_name : str Name of the microscope. - device_connection : str - Connection string for the remote focus device. - configuration : dict + device_connection : Any + Device connection object. + configuration : DictProxy Configuration dictionary. """ diff --git a/src/navigate/model/devices/remote_focus/equipment_solutions.py b/src/navigate/model/devices/remote_focus/equipment_solutions.py index 7f804cb7e..729dd767e 100644 --- a/src/navigate/model/devices/remote_focus/equipment_solutions.py +++ b/src/navigate/model/devices/remote_focus/equipment_solutions.py @@ -34,6 +34,8 @@ import time import serial import logging +from typing import Any +from multiprocessing.managers import DictProxy # Third Party Imports from navigate.tools.decorators import log_initialization @@ -60,16 +62,18 @@ class RemoteFocusEquipmentSolutions(RemoteFocusNI): sent. Once the character is received the next character can be processed. """ - def __init__(self, microscope_name, device_connection, configuration): + def __init__( + self, microscope_name: str, device_connection: Any, configuration: DictProxy + ) -> None: """Initialize the RemoteFocusEquipmentSolutions Class Parameters ---------- microscope_name : str Name of the microscope - device_connection : str - Name of the device connection - configuration : dict + device_connection : Any + Connection to the device + configuration : DictProxy Configuration dictionary """ super().__init__(microscope_name, device_connection, configuration) diff --git a/src/navigate/model/devices/remote_focus/ni.py b/src/navigate/model/devices/remote_focus/ni.py index 66e32a21f..ecaef62c0 100644 --- a/src/navigate/model/devices/remote_focus/ni.py +++ b/src/navigate/model/devices/remote_focus/ni.py @@ -32,6 +32,8 @@ # Standard Library Imports import logging +from typing import Any +from multiprocessing.managers import DictProxy # Third Party Imports @@ -48,16 +50,18 @@ class RemoteFocusNI(RemoteFocusBase): """RemoteFocusNI Class - Analog control of the remote focus device.""" - def __init__(self, microscope_name, device_connection, configuration): + def __init__( + self, microscope_name: str, device_connection: Any, configuration: DictProxy + ) -> None: """Initialize the RemoteFocusNI class. Parameters ---------- microscope_name : str The microscope name. - device_connection : object + device_connection : Any The device connection object. - configuration : dict + configuration : DictProxy The configuration dictionary. """ diff --git a/src/navigate/model/devices/remote_focus/synthetic.py b/src/navigate/model/devices/remote_focus/synthetic.py index 621ca9258..b6c3209ae 100644 --- a/src/navigate/model/devices/remote_focus/synthetic.py +++ b/src/navigate/model/devices/remote_focus/synthetic.py @@ -32,6 +32,8 @@ # Standard Library Imports import logging +from typing import Any +from multiprocessing.managers import DictProxy # Third Party Imports @@ -48,22 +50,25 @@ class SyntheticRemoteFocus(RemoteFocusBase): """SyntheticRemoteFocus Class""" - def __init__(self, microscope_name, device_connection, configuration): + def __init__( + self, microscope_name: str, device_connection: Any, configuration: DictProxy + ) -> None: """Initialize the SyntheticRemoteFocus class. Parameters ---------- microscope_name : str The microscope name. - device_connection : object + device_connection : Any The device connection object. - configuration : dict + configuration : DictProxy The device configuration. """ super().__init__(microscope_name, device_connection, configuration) pass - def move(self, readout_time, offset=None): + @staticmethod + def move(readout_time, offset=None): """Moves the remote focus. This method moves the remote focus. diff --git a/src/navigate/model/devices/stages/base.py b/src/navigate/model/devices/stages/base.py index fbf98b369..c15222582 100644 --- a/src/navigate/model/devices/stages/base.py +++ b/src/navigate/model/devices/stages/base.py @@ -31,7 +31,10 @@ # Standard Imports import logging -from multiprocessing.managers import ListProxy + +# from idlelib.debugger_r import DictProxy +from multiprocessing.managers import ListProxy, DictProxy +from typing import Any # Third Party Imports @@ -47,19 +50,25 @@ class StageBase: """Stage Parent Class""" - def __init__(self, microscope_name, device_connection, configuration, device_id=0): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: DictProxy, + device_id: int = 0, + ) -> None: """Initialize the stage. Parameters ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocessing.managers.DictProxy + configuration : DictProxy Global configuration of the microscope - device_id : int, optional - Device ID, by default 0 + device_id : int, Optional + Device ID. Default is 0. """ stage_configuration = configuration["configuration"]["microscopes"][ microscope_name diff --git a/src/navigate/model/devices/stages/sutter.py b/src/navigate/model/devices/stages/sutter.py index effdb5396..757287b10 100644 --- a/src/navigate/model/devices/stages/sutter.py +++ b/src/navigate/model/devices/stages/sutter.py @@ -32,6 +32,12 @@ # Standard Imports import logging import time + +# from idlelib.debugger_r import DictProxy +from multiprocessing.managers import DictProxy +from typing import Any + +# Third-Party Imports from serial import SerialException # Local Imports @@ -76,17 +82,25 @@ def build_MP285_connection(com_port: str, baud_rate: int, timeout=0.25) -> MP285 class SutterStage(StageBase): """SutterStage Class for MP-285.""" - def __init__(self, microscope_name, device_connection, configuration, device_id=0): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: DictProxy, + device_id: int = 0, + ) -> None: """Initialize the SutterStage. Parameters ---------- microscope_name : str Name of the microscope. - device_connection : MP285 - MP285 stage. - configuration : dict + device_connection : Any + MP285 stage connection. + configuration : DictProxy Configuration dictionary for the SutterStage. + device_id : int, Optional + Device ID for the SutterStage. Raises ------ diff --git a/src/navigate/tools/file_functions.py b/src/navigate/tools/file_functions.py index bdc7a840e..3433c3f06 100644 --- a/src/navigate/tools/file_functions.py +++ b/src/navigate/tools/file_functions.py @@ -37,6 +37,7 @@ import yaml from pathlib import Path import psutil +from typing import Union # Third party imports @@ -44,7 +45,7 @@ from navigate.tools.common_functions import copy_proxy_object -def get_ram_info(): +def get_ram_info() -> tuple: """Get computer RAM information. This function retrieves the total and available RAM on the computer. @@ -67,7 +68,7 @@ def get_ram_info(): return total_ram, available_ram -def create_save_path(saving_settings): +def create_save_path(saving_settings: dict) -> str: """Create path to save the data to. This function retrieves the user inputs from the popup save window. @@ -98,21 +99,22 @@ def create_save_path(saving_settings): label_string = label_string.replace(" ", "-") # Create the save directory on disk. - save_directory = os.path.join( - root_directory, - user_string, - tissue_string, - cell_type_string, - label_string, - date_string, + save_directory = str( + os.path.join( + root_directory, + user_string, + tissue_string, + cell_type_string, + label_string, + date_string, + ) ) os.makedirs(save_directory, exist_ok=True) - # Determine Number of Cells in Directory - # Cell1/Position1/1_CH00_000000.tif + + # Determine Number of Acquisitions in Directory prefix_num = len(prefix_string) cell_directories = list( - # filter(lambda v: v[:5] == "Cell_", os.listdir(save_directory)) filter(lambda v: v[:prefix_num] == prefix_string, os.listdir(save_directory)) ) if len(cell_directories) != 0: @@ -120,7 +122,6 @@ def create_save_path(saving_settings): cell_index = int(cell_directories[-1][prefix_num:]) + 1 else: cell_index = 1 - # cell_string = "Cell_" + str(cell_index).zfill(3) cell_string = prefix_string + str(cell_index).zfill(3) save_directory = os.path.join(save_directory, cell_string) @@ -133,19 +134,19 @@ def create_save_path(saving_settings): return save_directory -def load_yaml_file(file_path): +def load_yaml_file(file_path: str) -> Union[dict, None]: """Load YAML file from Disk Parameters ---------- - file_path : str/os.path + file_path : str String or path of the yaml file. Returns ------- config_data: dict/list/None - A dictionary/list of the yaml file content. - None: if the yaml file has error or not exist. + - A dictionary/list of the yaml file content. + - None: if the yaml file has error or not exist. """ file_path = Path(file_path) if not file_path.exists(): @@ -159,7 +160,9 @@ def load_yaml_file(file_path): return config_data -def save_yaml_file(file_directory, content_dict, filename="experiment.yml"): +def save_yaml_file( + file_directory: str, content_dict: dict, filename="experiment.yml" +) -> bool: """Same YAML file to Disk Parameters @@ -195,7 +198,7 @@ def save_yaml_file(file_directory, content_dict, filename="experiment.yml"): return True -def delete_folder(top): +def delete_folder(top: str) -> None: """Delete folder and all sub-folders. https://docs.python.org/3/library/os.html#os.walk From 058daf74cca32cc9df2c44ed191290533bbd5e26 Mon Sep 17 00:00:00 2001 From: Kevin Dean <42547789+AdvancedImagingUTSW@users.noreply.github.com> Date: Tue, 8 Oct 2024 08:23:28 -0500 Subject: [PATCH 4/5] Down the rabbit hole... Only significant change is in RemoteFocusEquipmentSolutions: num_waiting = self.serial.in_waiting Was inWaiting() Need to confirm that this is not a problem... --- .../model/data_sources/bdv_data_source.py | 6 +- .../model/data_sources/data_source.py | 31 ++++---- .../data_sources/pyramidal_data_source.py | 6 +- .../model/device_startup_functions.py | 78 ++++++++++--------- src/navigate/model/devices/camera/base.py | 24 +++--- .../model/devices/camera/hamamatsu.py | 66 ++++++++++------ .../model/devices/camera/photometrics.py | 14 ++-- src/navigate/model/devices/daq/base.py | 8 +- src/navigate/model/devices/daq/ni.py | 7 +- src/navigate/model/devices/daq/synthetic.py | 6 +- src/navigate/model/devices/galvo/base.py | 9 ++- src/navigate/model/devices/galvo/ni.py | 13 ++-- src/navigate/model/devices/galvo/synthetic.py | 9 +-- src/navigate/model/devices/lasers/base.py | 21 ++--- src/navigate/model/devices/lasers/ni.py | 7 +- .../model/devices/lasers/synthetic.py | 11 ++- .../model/devices/remote_focus/base.py | 17 ++-- .../remote_focus/equipment_solutions.py | 12 +-- src/navigate/model/devices/remote_focus/ni.py | 10 ++- .../model/devices/remote_focus/synthetic.py | 14 ++-- src/navigate/model/devices/shutter/base.py | 33 +++++--- src/navigate/model/devices/shutter/ni.py | 12 ++- .../model/devices/shutter/synthetic.py | 12 ++- src/navigate/model/devices/stages/asi.py | 31 +++++--- .../model/devices/stages/asi_MSTwoThousand.py | 26 +++++-- src/navigate/model/devices/stages/base.py | 15 ++-- src/navigate/model/devices/stages/ni.py | 20 +++-- src/navigate/model/devices/stages/pi.py | 37 +++++---- src/navigate/model/devices/stages/sutter.py | 17 ++-- src/navigate/model/devices/zoom/synthetic.py | 15 ++-- .../model/metadata_sources/metadata.py | 10 ++- src/navigate/model/microscope.py | 7 +- src/navigate/model/waveforms.py | 7 +- 33 files changed, 364 insertions(+), 247 deletions(-) diff --git a/src/navigate/model/data_sources/bdv_data_source.py b/src/navigate/model/data_sources/bdv_data_source.py index 55646c53e..27b50a3d6 100644 --- a/src/navigate/model/data_sources/bdv_data_source.py +++ b/src/navigate/model/data_sources/bdv_data_source.py @@ -31,8 +31,8 @@ # Standard Imports import os -from multiprocessing.managers import DictProxy import logging +from typing import Any, Dict # Third Party Imports import h5py @@ -119,13 +119,13 @@ def get_slice(self, x, y, c, z=0, t=0, p=0, subdiv=0) -> npt.ArrayLike: return self.image[setup][z, y, x] def set_metadata_from_configuration_experiment( - self, configuration: DictProxy, microscope_name: str = None + self, configuration: Dict[str, Any], microscope_name: str = None ) -> None: """Sets the metadata from according to the microscope configuration. Parameters ---------- - configuration : DictProxy + configuration : Dict[str, Any] The configuration experiment. microscope_name : str The microscope name diff --git a/src/navigate/model/data_sources/data_source.py b/src/navigate/model/data_sources/data_source.py index 216b04b02..eb7a30777 100644 --- a/src/navigate/model/data_sources/data_source.py +++ b/src/navigate/model/data_sources/data_source.py @@ -32,12 +32,12 @@ # Standard Library Imports import logging +from typing import Any, Dict # Third Party Imports import numpy.typing as npt # Local Imports -from multiprocessing.managers import DictProxy # Logger Setup p = __name__.split(".")[1] @@ -75,10 +75,10 @@ def __init__(self, file_name: str = "", mode: str = "w") -> None: #: npt.ArrayLike: Pointer to the metadata. self.metadata = None # Expect a metadata object - #: str: Mode to open the file in. Can be 'r' or 'w'. + # str: Mode to open the file in. Can be 'r' or 'w'. self._mode = None - #: bool: Has the data source been closed? + # bool: Has the data source been closed? self._closed = True #: int: Number of bits per pixel. @@ -99,17 +99,20 @@ def __init__(self, file_name: str = "", mode: str = "w") -> None: self.dc = 1 # step size between channels, should always be 1 #: int: Size of the data source in x dimension. + self.shape_x = 1 + #: int: Size of the data source in y dimension. + self.shape_y = 1 + #: int: Size of the data source in z dimension. + self.shape_z = 1 + #: int: Size of the data source in t dimension. + self.shape_t = 1 + #: int: Size of the data source in c dimension. - self.shape_x, self.shape_y, self.shape_z, self.shape_t, self.shape_c = ( - 1, - 1, - 1, - 1, - 1, - ) + self.shape_c = 1 + #: int: Number of positions in the data source. self.positions = 1 @@ -121,14 +124,14 @@ def __init__(self, file_name: str = "", mode: str = "w") -> None: @property def nbytes(self) -> int: - """Getter for the size of this data source in bytes." + """Getter for the size of this data source in bytes. Does not account for pyramidal data sources. That process is handled by the bdv_data_source class, which is a child of this class. Returns ------- - int + total_bytes : int Size of this data source in bytes. """ total_bits = ( @@ -216,13 +219,13 @@ def setup(self): pass def set_metadata_from_configuration_experiment( - self, configuration: DictProxy, microscope_name: str = None + self, configuration: Dict[str, Any], microscope_name: str = None ) -> None: """Sets the metadata from according to the microscope configuration. Parameters ---------- - configuration : DictProxy + configuration : Dict[str, Any] Configuration experiment. microscope_name : str The microscope name diff --git a/src/navigate/model/data_sources/pyramidal_data_source.py b/src/navigate/model/data_sources/pyramidal_data_source.py index ceffabc6b..3c121a85b 100644 --- a/src/navigate/model/data_sources/pyramidal_data_source.py +++ b/src/navigate/model/data_sources/pyramidal_data_source.py @@ -31,8 +31,8 @@ # POSSIBILITY OF SUCH DAMAGE. # Standard library imports -from multiprocessing.managers import DictProxy import logging +from typing import Any, Dict # Third-party imports import numpy as np @@ -154,13 +154,13 @@ def nbytes(self) -> int: ).sum() def set_metadata_from_configuration_experiment( - self, configuration: DictProxy, microscope_name: str = None + self, configuration: Dict[str, Any], microscope_name: str = None ) -> None: """Sets the metadata from according to the microscope configuration. Parameters ---------- - configuration : DictProxy + configuration : Dict[str, Any] The configuration experiment. microscope_name : str The microscope name diff --git a/src/navigate/model/device_startup_functions.py b/src/navigate/model/device_startup_functions.py index c13d2ea3e..e100c0973 100644 --- a/src/navigate/model/device_startup_functions.py +++ b/src/navigate/model/device_startup_functions.py @@ -34,8 +34,8 @@ import logging import time import importlib -from multiprocessing.managers import ListProxy, DictProxy -from typing import Callable, Tuple, Any, Type +from multiprocessing.managers import ListProxy +from typing import Callable, Tuple, Any, Type, Dict, Union, Optional # Third Party Imports @@ -182,7 +182,7 @@ def build_connection( def load_camera_connection( - configuration: DictProxy, camera_id: int = 0, is_synthetic: bool = False + configuration: Dict[str, Any], camera_id: int = 0, is_synthetic: bool = False ) -> Any: """Initialize the camera API class. @@ -191,7 +191,7 @@ def load_camera_connection( Parameters ---------- - configuration : DictProxy + configuration : Dict[str, Any] Global configuration of the microscope camera_id : int, Optional Device ID (0, 1...) @@ -248,7 +248,7 @@ def load_camera_connection( def start_camera( microscope_name: str, device_connection: Any, - configuration: DictProxy, + configuration: Dict[str, Any], is_synthetic: bool = False, plugin_devices=None, ) -> CameraBase: @@ -260,7 +260,7 @@ def start_camera( Name of microscope in configuration device_connection : Any Hardware device to connect to - configuration : DictProxy + configuration : Dict[str, Any] Global configuration of the microscope is_synthetic : bool, Optional Run synthetic version of hardware. Default is False. @@ -332,12 +332,12 @@ def start_camera( device_not_found(microscope_name, "camera", cam_type) -def load_mirror(configuration: DictProxy, is_synthetic: bool = False) -> Any: +def load_mirror(configuration: Dict[str, Any], is_synthetic: bool = False) -> Any: """Initializes the deformable mirror API. Parameters ---------- - configuration : DictProxy + configuration : Dict[str, Any] Global configuration of the microscope is_synthetic : bool, Optional Run synthetic version of hardware. Default is False. @@ -367,9 +367,9 @@ def load_mirror(configuration: DictProxy, is_synthetic: bool = False) -> Any: def start_mirror( microscope_name: str, device_connection: Any, - configuration: DictProxy, + configuration: Dict[str, Any], is_synthetic: bool = False, - plugin_devices={}, + plugin_devices: Union[Dict, None, Optional] = None, ) -> MirrorBase: """Initialize the mirror class. @@ -379,18 +379,20 @@ def start_mirror( Name of microscope in configuration device_connection : Any Hardware device to connect to - configuration : DictProxy + configuration : Dict[str, Any] Global configuration of the microscope is_synthetic : bool, Optional Run synthetic version of hardware. Default is False. - plugin_devices : dict - Dictionary of plugin devices + plugin_devices : dict, Optional + Dictionary of plugin devices. Default is None. Returns ------- mirror : Any Mirror class. """ + if plugin_devices is None: + plugin_devices = {} if device_connection is None or is_synthetic: mirror_type = "SyntheticMirror" @@ -419,7 +421,7 @@ def start_mirror( def load_stages( - configuration: DictProxy, is_synthetic: bool = False, plugin_devices=None + configuration: Dict[str, Any], is_synthetic: bool = False, plugin_devices=None ) -> Any: """Initialize the stage API. @@ -428,7 +430,7 @@ def load_stages( Parameters ---------- - configuration : DictProxy + configuration : Dict[str, Any] Global configuration of the microscope is_synthetic : bool, Optional Run synthetic version of hardware. Default is False. @@ -641,7 +643,7 @@ def load_stages( def start_stage( microscope_name: str, device_connection: Any, - configuration: DictProxy, + configuration: Dict[str, Any], id: int = 0, is_synthetic: bool = False, plugin_devices=None, @@ -657,7 +659,7 @@ def start_stage( Name of microscope in configuration device_connection : Any Hardware device to connect to - configuration : DictProxy + configuration : Dict[str, Any] Global configuration of the microscope id : int, Optional ID of the stage. Default is 0. @@ -757,7 +759,7 @@ def start_stage( def load_zoom_connection( - configuration: DictProxy, is_synthetic: bool = False, plugin_devices=None + configuration: Dict[str, Any], is_synthetic: bool = False, plugin_devices=None ) -> Any: """Initializes the Zoom class on a dedicated thread. @@ -766,7 +768,7 @@ def load_zoom_connection( Parameters ---------- - configuration : DictProxy + configuration : Dict[str, Any] Global configuration of the microscope is_synthetic : bool Run synthetic version of hardware? @@ -816,7 +818,7 @@ def load_zoom_connection( def start_zoom( microscope_name: str, device_connection: Any, - configuration: DictProxy, + configuration: Dict[str, Any], is_synthetic: bool = False, plugin_devices=None, ) -> ZoomBase: @@ -830,7 +832,7 @@ def start_zoom( Name of microscope in configuration device_connection : Any Hardware device to connect to - configuration : DictProxy + configuration : Dict[str, Any] Global configuration of the microscope is_synthetic : bool, Optional Run synthetic version of hardware. Default is False. @@ -893,7 +895,7 @@ def start_zoom( def load_filter_wheel_connection( - device_info: DictProxy, is_synthetic=False, plugin_devices=None + device_info: Dict[str, Any], is_synthetic=False, plugin_devices=None ) -> Any: """Initializes the Filter Wheel class on a dedicated thread. @@ -902,7 +904,7 @@ def load_filter_wheel_connection( Parameters ---------- - device_info : DictProxy + device_info : Dict[str, Any] filter wheel device configuration is_synthetic : bool, Optional Run synthetic version of hardware, Default is False. @@ -982,7 +984,7 @@ def load_filter_wheel_connection( def start_filter_wheel( microscope_name: str, device_connection: Any, - configuration: DictProxy, + configuration: Dict[str, Any], id: int = 0, is_synthetic: bool = False, plugin_devices=None, @@ -997,7 +999,7 @@ def start_filter_wheel( Name of microscope in configuration device_connection : Any Device connection - configuration : DictProxy + configuration : Dict[str, Any] Global configuration of the microscope id : int, Optional Index of filter_wheel in the configuration dictionary. Default is 0. @@ -1079,7 +1081,7 @@ def start_filter_wheel( device_not_found(microscope_name, "filter_wheel", device_type) -def start_daq(configuration: DictProxy, is_synthetic: bool = False) -> DAQBase: +def start_daq(configuration: Dict[str, Any], is_synthetic: bool = False) -> DAQBase: """Initializes the data acquisition (DAQ) class on a dedicated thread. Load daq information from the configuration file. Proper daq types include NI and @@ -1087,7 +1089,7 @@ def start_daq(configuration: DictProxy, is_synthetic: bool = False) -> DAQBase: Parameters ---------- - configuration : DictProxy + configuration : Dict[str, Any] Global configuration of the microscope is_synthetic : bool, Optional Run synthetic version of hardware. Default is False. @@ -1120,7 +1122,7 @@ def start_daq(configuration: DictProxy, is_synthetic: bool = False) -> DAQBase: def start_shutter( microscope_name: str, device_connection: Any, - configuration: DictProxy, + configuration: Dict[str, Any], is_synthetic: bool = False, plugin_devices: dict = None, ) -> ShutterBase: @@ -1136,7 +1138,7 @@ def start_shutter( Name of microscope in configuration device_connection : Any Hardware device to connect to - configuration : DictProxy + configuration : Dict[str, Any] Global configuration of the microscope is_synthetic : bool, Optional Run synthetic version of hardware. Default is False. @@ -1196,7 +1198,7 @@ def start_shutter( def start_lasers( microscope_name: str, device_connection: Any, - configuration: DictProxy, + configuration: Dict[str, Any], id: int = 0, is_synthetic: bool = False, plugin_devices: dict = None, @@ -1212,7 +1214,7 @@ def start_lasers( Name of microscope in configuration device_connection : Any Hardware device to connect to - configuration : DictProxy + configuration : Dict[str, Any] Global configuration of the microscope id : int, Optional Index of laser in laser list in configuration dictionary. Default is 0. @@ -1274,7 +1276,7 @@ def start_lasers( def start_remote_focus_device( microscope_name: str, device_connection: Any, - configuration: DictProxy, + configuration: Dict[str, Any], is_synthetic: bool = False, plugin_devices=None, ) -> RemoteFocusBase: @@ -1289,7 +1291,7 @@ def start_remote_focus_device( Name of microscope in configuration device_connection : Any Hardware device to connect to - configuration : DictProxy + configuration : Dict[str, Any] Global configuration of the microscope is_synthetic : bool, Optional Run synthetic version of hardware. Default is False. @@ -1358,7 +1360,7 @@ def start_remote_focus_device( def start_galvo( microscope_name: str, device_connection: Any, - configuration: DictProxy, + configuration: Dict[str, Any], id: int = 0, is_synthetic: bool = False, plugin_devices=None, @@ -1373,7 +1375,7 @@ def start_galvo( Name of microscope in configuration device_connection : Any Hardware device to connect to - configuration : DictProxy + configuration : Dict[str, Any] Global configuration of the microscope id : int, Optional Index of galvo in the configuration dictionary. Default is 0. @@ -1458,13 +1460,13 @@ def device_not_found(*args: Any) -> None: def load_devices( - configuration: DictProxy, is_synthetic=False, plugin_devices=None + configuration: Dict[str, Any], is_synthetic=False, plugin_devices=None ) -> dict: """Load devices from configuration. Parameters ---------- - configuration : DictProxy + configuration : Dict[str, Any] Configuration dictionary is_synthetic : bool Run synthetic version of hardware? @@ -1474,7 +1476,7 @@ def load_devices( Returns ------- devices : dict - Dictionary of devcies + Dictionary of devices """ if plugin_devices is None: diff --git a/src/navigate/model/devices/camera/base.py b/src/navigate/model/devices/camera/base.py index a237b7867..eab68b080 100644 --- a/src/navigate/model/devices/camera/base.py +++ b/src/navigate/model/devices/camera/base.py @@ -33,6 +33,7 @@ # Standard Library Imports import logging import os +from typing import Any, Dict # Third Party Imports import tifffile @@ -50,16 +51,21 @@ class CameraBase: """CameraBase - Parent camera class.""" - def __init__(self, microscope_name, device_connection, configuration): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], + ): """Initialize CameraBase class. Parameters ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocessing.managers.DictProxy + configuration : Dict[str, Any] Global configuration of the microscope Raises @@ -155,8 +161,8 @@ def get_offset_variance_maps(self): If offset or variance map is not found. """ serial_number = self.camera_parameters["hardware"]["serial_number"] + map_path = os.path.join(get_navigate_path(), "camera_maps") try: - map_path = os.path.join(get_navigate_path(), "camera_maps") self._offset = tifffile.imread( os.path.join(map_path, f"{serial_number}_off.tiff") ) @@ -164,9 +170,7 @@ def get_offset_variance_maps(self): os.path.join(map_path, f"{serial_number}_var.tiff") ) except FileNotFoundError: - logger.info( - f"{str(self)}, Offset or variance map not found in {map_path}" - ) + logger.info(f"{str(self)}, Offset or variance map not found in {map_path}") self._offset, self._variance = None, None return self._offset, self._variance @@ -197,7 +201,7 @@ def variance(self): self.get_offset_variance_maps() return self._variance - def set_readout_direction(self, mode): + def set_readout_direction(self, mode) -> None: """Set HamamatsuOrca readout direction. Parameters @@ -239,11 +243,11 @@ def calculate_light_sheet_exposure_time( exposure_time = camera_line_interval * shutter_width return exposure_time, camera_line_interval, full_chip_exposure_time - def close_camera(self): + def close_camera(self) -> None: """Close camera.""" pass - def get_line_interval(self): + def get_line_interval(self) -> float: """Return stored camera line interval. Returns diff --git a/src/navigate/model/devices/camera/hamamatsu.py b/src/navigate/model/devices/camera/hamamatsu.py index 36089764f..f71afbbcf 100644 --- a/src/navigate/model/devices/camera/hamamatsu.py +++ b/src/navigate/model/devices/camera/hamamatsu.py @@ -32,6 +32,7 @@ # Standard Library Imports import logging +from typing import Any, Dict # Third Party Imports @@ -52,16 +53,21 @@ class HamamatsuBase(CameraBase): This includes the ORCA Flash 4.0, Fusion, Lightning, and Fire. """ - def __init__(self, microscope_name, device_connection, configuration): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], + ) -> None: """Initialize HamamatsuOrca class. Parameters ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocessing.managers.DictProxy + configuration : Dict[str, Any] Global configuration of the microscope """ super().__init__(microscope_name, device_connection, configuration) @@ -252,18 +258,12 @@ def set_readout_direction(self, mode): logger.debug("Camera readout direction not supported") def calculate_readout_time(self): - """Calculate duration of time needed to readout an image. - - Calculates the readout time and maximum frame rate according to the camera - configuration settings. - Assumes model C13440 with Camera Link communication from Hamamatsu. - Currently pulling values directly from the camera. + """Get the duration of time needed to read out an image. Returns ------- readout_time : float - Duration of time needed to readout an image. - + Duration of time needed to read out an image. """ readout_time = self.camera_controller.get_property_value("readout_time") @@ -446,16 +446,21 @@ def get_new_frame(self): class HamamatsuOrcaLightning(HamamatsuBase): """HamamatsuOrcaLightning camera class.""" - def __init__(self, microscope_name, device_connection, configuration): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], + ) -> None: """Initialize HamamatsuOrcaLightning class. Parameters ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocessing.managers.DictProxy + configuration : Dict[str, Any] Global configuration of the microscope """ HamamatsuBase.__init__(self, microscope_name, device_connection, configuration) @@ -514,16 +519,21 @@ def calculate_light_sheet_exposure_time( @log_initialization class HamamatsuOrcaFire(HamamatsuBase): - def __init__(self, microscope_name, device_connection, configuration): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], + ) -> None: """Initialize HamamatsuOrcaFire class. Parameters ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocessing.managers.DictProxy + configuration : Dict[str, Any] Global configuration of the microscope """ HamamatsuBase.__init__(self, microscope_name, device_connection, configuration) @@ -597,7 +607,12 @@ def calculate_light_sheet_exposure_time( @log_initialization class HamamatsuOrca(HamamatsuBase): - def __init__(self, microscope_name, device_connection, configuration): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], + ) -> None: """Initialize HamamatsuOrca class. This is for controlling the Orca Flash 4.0. @@ -606,9 +621,9 @@ def __init__(self, microscope_name, device_connection, configuration): ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocessing.managers.DictProxy + configuration : Dict[str, Any] Global configuration of the microscope """ HamamatsuBase.__init__(self, microscope_name, device_connection, configuration) @@ -674,16 +689,21 @@ def calculate_light_sheet_exposure_time( class HamamatsuOrcaFusion(HamamatsuBase): """HamamatsuOrcaFusion camera class.""" - def __init__(self, microscope_name, device_connection, configuration): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], + ) -> None: """Initialize HamamatsuOrcaFusion class. Parameters ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocessing.managers.DictProxy + configuration : Dict[str, Any] Global configuration of the microscope """ HamamatsuBase.__init__(self, microscope_name, device_connection, configuration) diff --git a/src/navigate/model/devices/camera/photometrics.py b/src/navigate/model/devices/camera/photometrics.py index c5896346d..d46fa83ca 100644 --- a/src/navigate/model/devices/camera/photometrics.py +++ b/src/navigate/model/devices/camera/photometrics.py @@ -32,6 +32,7 @@ # Standard Library Imports import logging +from typing import Any, Dict # Third Party Imports from ctypes import * # noqa @@ -91,19 +92,22 @@ class PhotometricsBase(CameraBase): python setup.py install """ - def __init__(self, microscope_name, device_connection, configuration): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], + ) -> None: """Initialize the Photometrics class. Parameters --------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to. will be saved in camera_controller - configuration : multiprocessing.managers.DictProxy + configuration : Dict[str, Any] Global configuration of the microscope - ------- - """ super().__init__(microscope_name, device_connection, configuration) diff --git a/src/navigate/model/devices/daq/base.py b/src/navigate/model/devices/daq/base.py index 405ab59dd..e3c934571 100644 --- a/src/navigate/model/devices/daq/base.py +++ b/src/navigate/model/devices/daq/base.py @@ -32,7 +32,7 @@ # Standard Imports import logging -from multiprocessing.managers import DictProxy +from typing import Any, Dict # Third Party Imports @@ -49,12 +49,12 @@ class DAQBase: """DAQBase - Parent class for Data Acquisition (DAQ) classes.""" - def __init__(self, configuration: DictProxy): + def __init__(self, configuration: Dict[str, Any]) -> None: """Initializes the DAQBase class. Parameters ---------- - configuration : DictProxy + configuration : Dict[str, Any] Dictionary of configuration parameters """ @@ -150,7 +150,7 @@ def calculate_all_waveforms(self, microscope_name, exposure_times, sweep_times): return self.waveform_dict - def enable_microscope(self, microscope_name): + def enable_microscope(self, microscope_name: str) -> None: """Enables the microscope. Parameters diff --git a/src/navigate/model/devices/daq/ni.py b/src/navigate/model/devices/daq/ni.py index b9c490471..9034d7836 100644 --- a/src/navigate/model/devices/daq/ni.py +++ b/src/navigate/model/devices/daq/ni.py @@ -32,11 +32,10 @@ # Standard Imports import logging -import multiprocessing.managers from threading import Lock import traceback import time -from typing import Union +from typing import Union, Dict, Any # Third Party Imports import nidaqmx @@ -58,12 +57,12 @@ class NIDAQ(DAQBase): """NIDAQ class for Control of NI Data Acquisition Cards.""" - def __init__(self, configuration: multiprocessing.managers.DictProxy) -> None: + def __init__(self, configuration: Dict[str, Any]) -> None: """Initialize NIDAQ class. Parameters ---------- - configuration : multiprocessing.managers.DictProxy + configuration : Dict[str, Any] Configuration dictionary. """ super().__init__(configuration) diff --git a/src/navigate/model/devices/daq/synthetic.py b/src/navigate/model/devices/daq/synthetic.py index b21b6d48e..fc73a4eef 100644 --- a/src/navigate/model/devices/daq/synthetic.py +++ b/src/navigate/model/devices/daq/synthetic.py @@ -34,9 +34,9 @@ import logging import time from threading import Lock +from typing import Any, Dict # Third Party Imports -from multiprocessing.managers import DictProxy # Local Imports from navigate.model.devices.daq.base import DAQBase @@ -51,12 +51,12 @@ class SyntheticDAQ(DAQBase): """SyntheticDAQ class for Data Acquisition (DAQ).""" - def __init__(self, configuration: DictProxy) -> None: + def __init__(self, configuration: Dict[str, Any]) -> None: """Initialize the Synthetic DAQ. Parameters ---------- - configuration : DictProxy + configuration : Dict[str, Any] Configuration dictionary. """ super().__init__(configuration) diff --git a/src/navigate/model/devices/galvo/base.py b/src/navigate/model/devices/galvo/base.py index 0544940aa..8bff84692 100644 --- a/src/navigate/model/devices/galvo/base.py +++ b/src/navigate/model/devices/galvo/base.py @@ -32,8 +32,7 @@ # Standard Library Imports import logging -from typing import Any -from multiprocessing.managers import DictProxy +from typing import Any, Dict # Third Party Imports @@ -54,7 +53,7 @@ def __init__( self, microscope_name: str, device_connection: Any, - configuration: DictProxy, + configuration: Dict[str, Any], galvo_id: int = 0, ) -> None: """Initialize the GalvoBase class. @@ -65,11 +64,13 @@ def __init__( Name of the microscope. device_connection : Any Device connection. - configuration : DictProxy + configuration : Dict[str, Any] Dictionary of configuration parameters. galvo_id : int Galvo ID. Default is 0. """ + #: Any: Device connection. + self.device_connection = device_connection #: dict: Dictionary of microscope configuration parameters. self.configuration = configuration diff --git a/src/navigate/model/devices/galvo/ni.py b/src/navigate/model/devices/galvo/ni.py index 386d3769d..40cfe71b3 100644 --- a/src/navigate/model/devices/galvo/ni.py +++ b/src/navigate/model/devices/galvo/ni.py @@ -32,8 +32,7 @@ # Standard Library Imports import logging -from typing import Any -from multiprocessing.managers import DictProxy +from typing import Any, Dict # Third Party Imports import nidaqmx @@ -55,7 +54,7 @@ def __init__( self, microscope_name: str, device_connection: Any, - configuration: DictProxy, + configuration: Dict[str, Any], galvo_id: int = 0, ) -> None: """Initialize the GalvoNI class. @@ -66,7 +65,7 @@ def __init__( Name of the microscope. device_connection : Any Connection to the NI DAQ device. - configuration : DictProxy + configuration : Dict[str, Any] Dictionary of configuration parameters. galvo_id : int Galvo ID. Default is 0. @@ -90,11 +89,11 @@ def __init__( #: obj: NI DAQ device connection. self.daq = device_connection - def __str__(self): + def __str__(self) -> str: """Return string representation of the GalvoNI.""" return "GalvoNI" - def adjust(self, exposure_times, sweep_times): + def adjust(self, exposure_times, sweep_times) -> Dict[str, Any]: """Adjust the galvo to the readout time Parameters @@ -117,7 +116,7 @@ def adjust(self, exposure_times, sweep_times): } return waveform_dict - def turn_off(self): + def turn_off(self) -> None: """Turn off the galvo. Turns off the galvo. NOTE: This will only work if there isn't another task bound to this channel. This should only be called in microscope.terminate(). diff --git a/src/navigate/model/devices/galvo/synthetic.py b/src/navigate/model/devices/galvo/synthetic.py index 8065d6941..57ed6dbf1 100644 --- a/src/navigate/model/devices/galvo/synthetic.py +++ b/src/navigate/model/devices/galvo/synthetic.py @@ -32,8 +32,7 @@ # Standard Library Imports import logging -from typing import Any -from multiprocessing.managers import DictProxy +from typing import Any, Dict # Third Party Imports @@ -54,7 +53,7 @@ def __init__( self, microscope_name: str, device_connection: Any, - configuration: DictProxy, + configuration: Dict[str, Any], galvo_id: int = 0, ) -> None: """Initialize the SyntheticGalvo class. @@ -65,7 +64,7 @@ def __init__( Name of the microscope. device_connection : Any Device connection. - configuration : DictProxy + configuration : Dict[str, Any] Dictionary of configuration parameters. galvo_id : int Galvo ID. Default is 0. @@ -84,6 +83,6 @@ def __init__( #: int: Galvo ID. self.galvo_id = galvo_id - def __str__(self): + def __str__(self) -> str: """Return string representation of the GalvoNI.""" return "SyntheticGalvo" diff --git a/src/navigate/model/devices/lasers/base.py b/src/navigate/model/devices/lasers/base.py index 02da3be5d..c2a145e84 100644 --- a/src/navigate/model/devices/lasers/base.py +++ b/src/navigate/model/devices/lasers/base.py @@ -32,8 +32,7 @@ # Standard Library Imports import logging -from typing import Any -from multiprocessing.managers import DictProxy +from typing import Any, Dict # Third Party Imports @@ -53,7 +52,7 @@ def __init__( self, microscope_name: str, device_connection: Any, - configuration: DictProxy, + configuration: Dict[str, Any], laser_id: int, ) -> None: """Initialize Laser Base Class @@ -64,11 +63,13 @@ def __init__( Name of the microscope device_connection : Any Communication instance with the device. - configuration : DictProxy + configuration : Dict[str, Any] Configuration dictionary laser_id : int Laser ID """ + #: Any: Communication instance with the device + self.device_connection = device_connection #: dict: Configuration dictionary self.configuration = configuration @@ -81,11 +82,11 @@ def __init__( microscope_name ]["lasers"][laser_id] - def __str__(self): + def __str__(self) -> str: """Return string representation of the class""" return "LaserBase" - def set_power(self, laser_intensity): + def set_power(self, laser_intensity: int) -> None: """Set laser power Parameters @@ -95,20 +96,20 @@ def set_power(self, laser_intensity): """ pass - def turn_on(self): + def turn_on(self) -> None: """Turn on the laser""" pass - def turn_off(self): + def turn_off(self) -> None: """Turn off the laser""" pass - def close(self): + def close(self) -> None: """ Close the laser before exit. """ pass - def initialize_laser(self): + def initialize_laser(self) -> None: """Initialize lasers.""" pass diff --git a/src/navigate/model/devices/lasers/ni.py b/src/navigate/model/devices/lasers/ni.py index d6512d2c4..5a5f116c6 100644 --- a/src/navigate/model/devices/lasers/ni.py +++ b/src/navigate/model/devices/lasers/ni.py @@ -32,8 +32,7 @@ # Standard Library Imports import logging -from typing import Any -from multiprocessing.managers import DictProxy +from typing import Any, Dict # Third Party Imports import nidaqmx @@ -60,7 +59,7 @@ def __init__( self, microscope_name: str, device_connection: Any, - configuration: DictProxy, + configuration: Dict[str, Any], laser_id: int, ) -> None: """Initialize the LaserNI class. @@ -71,7 +70,7 @@ def __init__( The microscope name. device_connection : Any The device connection object. - configuration : DictProxy + configuration : Dict[str, Any] The device configuration. laser_id : int The laser id. diff --git a/src/navigate/model/devices/lasers/synthetic.py b/src/navigate/model/devices/lasers/synthetic.py index c89fc4e99..554b8093b 100644 --- a/src/navigate/model/devices/lasers/synthetic.py +++ b/src/navigate/model/devices/lasers/synthetic.py @@ -32,8 +32,7 @@ # Standard Library Imports import logging -from typing import Any -from multiprocessing.managers import DictProxy +from typing import Any, Dict # Third Party Imports @@ -54,7 +53,7 @@ def __init__( self, microscope_name: str, device_connection: Any, - configuration: DictProxy, + configuration: Dict[str, Any], laser_id: int, ) -> None: """Initialize the SyntheticLaser class. @@ -65,18 +64,18 @@ def __init__( The microscope name. device_connection : Any The device connection object. - configuration : DictProxy + configuration : Dict[str, Any] The device configuration. laser_id : int The laser ID. """ super().__init__(microscope_name, device_connection, configuration, laser_id) - def close(self): + def close(self) -> None: """Close the port before exit.""" pass - def initialize_laser(self): + def initialize_laser(self) -> None: """ Initialize lasers. """ diff --git a/src/navigate/model/devices/remote_focus/base.py b/src/navigate/model/devices/remote_focus/base.py index 323165b66..4a4df7c24 100644 --- a/src/navigate/model/devices/remote_focus/base.py +++ b/src/navigate/model/devices/remote_focus/base.py @@ -32,8 +32,7 @@ # Standard Library Imports import logging -from typing import Any -from multiprocessing.managers import DictProxy +from typing import Any, Dict # Third Party Imports @@ -55,7 +54,10 @@ class RemoteFocusBase: """RemoteFocusBase Class - Parent class for Remote Focusing Device.""" def __init__( - self, microscope_name: str, device_connection: Any, configuration: DictProxy + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], ) -> None: """Initializes the RemoteFocusBase Class. @@ -65,10 +67,13 @@ def __init__( Name of the microscope. device_connection : Any Device connection object. - configuration : DictProxy + configuration : Dict[str, Any] Configuration dictionary. """ + #: Any: Device connection object. + self.device_connection = device_connection + #: dict: Configuration dictionary. self.configuration = configuration @@ -123,6 +128,8 @@ def adjust(self, exposure_times, sweep_times, offset=None): Dictionary of exposure times for each selected channel sweep_times : dict Dictionary of sweep times for each selected channel + offset : float, optional + Offset value for the remote focus waveform, by default None Returns ------- @@ -241,7 +248,7 @@ def adjust(self, exposure_times, sweep_times, offset=None): percent_smoothing=percent_smoothing, )[:samples] - # Clip any values outside of the hardware limits + # Clip any values outside the hardware limits self.waveform_dict[channel_key][ self.waveform_dict[channel_key] > self.remote_focus_max_voltage ] = self.remote_focus_max_voltage diff --git a/src/navigate/model/devices/remote_focus/equipment_solutions.py b/src/navigate/model/devices/remote_focus/equipment_solutions.py index 729dd767e..333de840d 100644 --- a/src/navigate/model/devices/remote_focus/equipment_solutions.py +++ b/src/navigate/model/devices/remote_focus/equipment_solutions.py @@ -34,8 +34,7 @@ import time import serial import logging -from typing import Any -from multiprocessing.managers import DictProxy +from typing import Any, Dict # Third Party Imports from navigate.tools.decorators import log_initialization @@ -63,7 +62,10 @@ class RemoteFocusEquipmentSolutions(RemoteFocusNI): """ def __init__( - self, microscope_name: str, device_connection: Any, configuration: DictProxy + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], ) -> None: """Initialize the RemoteFocusEquipmentSolutions Class @@ -73,7 +75,7 @@ def __init__( Name of the microscope device_connection : Any Connection to the device - configuration : DictProxy + configuration : Dict[str, Any] Configuration dictionary """ super().__init__(microscope_name, device_connection, configuration) @@ -171,7 +173,7 @@ def read_bytes(self, num_bytes: int) -> bytes: """ for i in range(100): - num_waiting = self.serial.inWaiting() + num_waiting = self.serial.in_waiting if num_waiting == num_bytes: break time.sleep(0.02) diff --git a/src/navigate/model/devices/remote_focus/ni.py b/src/navigate/model/devices/remote_focus/ni.py index ecaef62c0..0d2ddbaca 100644 --- a/src/navigate/model/devices/remote_focus/ni.py +++ b/src/navigate/model/devices/remote_focus/ni.py @@ -32,8 +32,7 @@ # Standard Library Imports import logging -from typing import Any -from multiprocessing.managers import DictProxy +from typing import Any, Dict # Third Party Imports @@ -51,7 +50,10 @@ class RemoteFocusNI(RemoteFocusBase): """RemoteFocusNI Class - Analog control of the remote focus device.""" def __init__( - self, microscope_name: str, device_connection: Any, configuration: DictProxy + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], ) -> None: """Initialize the RemoteFocusNI class. @@ -61,7 +63,7 @@ def __init__( The microscope name. device_connection : Any The device connection object. - configuration : DictProxy + configuration : Dict[str, Any] The configuration dictionary. """ diff --git a/src/navigate/model/devices/remote_focus/synthetic.py b/src/navigate/model/devices/remote_focus/synthetic.py index b6c3209ae..e2a32b4b8 100644 --- a/src/navigate/model/devices/remote_focus/synthetic.py +++ b/src/navigate/model/devices/remote_focus/synthetic.py @@ -28,12 +28,11 @@ # IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE # POSSIBILITY OF SUCH DAMAGE. -# + # Standard Library Imports import logging -from typing import Any -from multiprocessing.managers import DictProxy +from typing import Any, Dict # Third Party Imports @@ -51,7 +50,10 @@ class SyntheticRemoteFocus(RemoteFocusBase): """SyntheticRemoteFocus Class""" def __init__( - self, microscope_name: str, device_connection: Any, configuration: DictProxy + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], ) -> None: """Initialize the SyntheticRemoteFocus class. @@ -61,7 +63,7 @@ def __init__( The microscope name. device_connection : Any The device connection object. - configuration : DictProxy + configuration : Dict[str, Any] The device configuration. """ super().__init__(microscope_name, device_connection, configuration) @@ -80,4 +82,4 @@ def move(readout_time, offset=None): offset : float The offset of the signal in volts. """ - logger.debug(f"move remote focus offset: {offset}") + logger.debug(f"Remote focus offset and readout time: {offset}, {readout_time}") diff --git a/src/navigate/model/devices/shutter/base.py b/src/navigate/model/devices/shutter/base.py index 259030d4e..7149d65df 100644 --- a/src/navigate/model/devices/shutter/base.py +++ b/src/navigate/model/devices/shutter/base.py @@ -32,6 +32,7 @@ # Standard Library Imports import logging +from typing import Any, Dict # Third Party Imports @@ -47,44 +48,58 @@ class ShutterBase: """ShutterBase Class - Parent class for the laser shutters.""" - def __init__(self, microscope_name, device_connection, configuration): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], + ) -> None: """Initialize the Shutter. Parameters ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocesing.managers.DictProxy + configuration : Dict[str, Any] Global configuration of the microscope """ #: bool: Shutter state self.shutter_state = False - def __str__(self): + #: str: Name of the microscope + self.microscope_name = microscope_name + + #: Any: Hardware device to connect to + self.device_connection = device_connection + + #: Dict[str, Any]: Global configuration of the microscope + self.configuration = configuration + + def __str__(self) -> str: """Return the string representation of the Shutter.""" return "ShutterBase" - def __del__(self): + def __del__(self) -> None: """Close the Shutter at exit.""" pass - def open_shutter(self): + def open_shutter(self) -> None: """Open the Shutter.""" self.shutter_state = True - def close_shutter(self): + def close_shutter(self) -> None: """Close the Shutter.""" self.shutter_state = False @property - def state(self): + def state(self) -> bool: """Get the Shutter state. Returns ------- bool - Shutter state + Shutter state. True if open, False if closed. """ return self.shutter_state diff --git a/src/navigate/model/devices/shutter/ni.py b/src/navigate/model/devices/shutter/ni.py index a4335baa7..5021f64ce 100644 --- a/src/navigate/model/devices/shutter/ni.py +++ b/src/navigate/model/devices/shutter/ni.py @@ -32,6 +32,7 @@ # Standard Library Imports import logging +from typing import Any, Dict # Third Party Imports import nidaqmx @@ -55,16 +56,21 @@ class ShutterTTL(ShutterBase): powered down. """ - def __init__(self, microscope_name, device_connection, configuration): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], + ) -> None: """Initialize the ShutterTTL. Parameters ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocessing.managers.DictProxy + configuration : Dict[str, Any] Global configuration of the microscope """ super().__init__(microscope_name, device_connection, configuration) diff --git a/src/navigate/model/devices/shutter/synthetic.py b/src/navigate/model/devices/shutter/synthetic.py index fdbdc218f..f184a97ee 100644 --- a/src/navigate/model/devices/shutter/synthetic.py +++ b/src/navigate/model/devices/shutter/synthetic.py @@ -33,6 +33,7 @@ # Standard Library Imports import logging +from typing import Any, Dict # Third Party Imports @@ -49,16 +50,21 @@ class SyntheticShutter(ShutterBase): """SyntheticShutter Class - Triggering for shutters delivered from synthetically.""" - def __init__(self, microscope_name, device_connection, configuration): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], + ) -> None: """Initialize the SyntheticShutter. Parameters ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocessing.managers.DictProxy + configuration : Dict[str, Any] Global configuration of the microscope """ super().__init__(microscope_name, device_connection, configuration) diff --git a/src/navigate/model/devices/stages/asi.py b/src/navigate/model/devices/stages/asi.py index 1f05afa79..c66f7a8ac 100644 --- a/src/navigate/model/devices/stages/asi.py +++ b/src/navigate/model/devices/stages/asi.py @@ -32,6 +32,7 @@ # Standard Imports import logging import time +from typing import Any, Dict, Optional, Union # Third Party Imports @@ -78,9 +79,9 @@ def build_ASI_Stage_connection(com_port, baud_rate=115200): class ASIStage(StageBase): """Applied Scientific Instrumentation (ASI) Stage Class - ASI Documentation: http://asiimaging.com/docs/products/serial_commands + ASI Documentation: https://asiimaging.com/docs/products/serial_commands - ASI Quick Start Guide: http://asiimaging.com/docs/command_quick_start + ASI Quick Start Guide: https://asiimaging.com/docs/command_quick_start Note ---- @@ -93,17 +94,25 @@ class ASIStage(StageBase): change both stilt positions simultaneously. """ - def __init__(self, microscope_name, device_connection, configuration, device_id=0): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], + device_id: Union[int, Optional] = 0, + ): """Initialize the ASI Stage connection. Parameters ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocessing.managers.DictProxy + configuration : Dict[str, Any] Global configuration of the microscope + device_id : Union[int, Optional] + Device ID for the stage, defaults to 0 """ super().__init__(microscope_name, device_connection, configuration, device_id) @@ -426,8 +435,9 @@ def scanr(self, start_position_mm, end_position_mm, enc_divide, axis="z"): start_position_mm, end_position_mm, enc_divide, axis ) except TigerException as e: - logger.exception(f"TigerException: {e}") - print(logger.exception()) + error_statement = f"TigerException: {e}" + logger.exception(error_statement) + print(error_statement) return False except KeyError as e: logger.exception(f"ASI Stage - KeyError in scanr: {e}") @@ -449,7 +459,7 @@ def scanv( scan start position end_position_mm: float scan end position - number of lines: int + number_of_lines: int number of steps. overshoot: float overshoot_time ms @@ -467,8 +477,9 @@ def scanv( start_position_mm, end_position_mm, number_of_lines, overshoot, axis ) except TigerException as e: - logger.exception(f"TigerException: {e}") - print(logger.exception()) + error_statement = f"TigerException: {e}" + logger.exception(error_statement) + print(error_statement) return False except KeyError as e: logger.exception(f"ASI Stage - KeyError in scanr: {e}") diff --git a/src/navigate/model/devices/stages/asi_MSTwoThousand.py b/src/navigate/model/devices/stages/asi_MSTwoThousand.py index 58670f26d..83bf3c1df 100644 --- a/src/navigate/model/devices/stages/asi_MSTwoThousand.py +++ b/src/navigate/model/devices/stages/asi_MSTwoThousand.py @@ -32,6 +32,7 @@ # Standard Imports import logging import time +from typing import Any, Dict, Optional, Union # Third Party Imports @@ -78,9 +79,9 @@ def build_ASI_Stage_connection(com_port, baud_rate=115200): class ASIStage(StageBase): """Applied Scientific Instrumentation (ASI) Stage Class - ASI Documentation: http://asiimaging.com/docs/products/serial_commands + ASI Documentation: https://asiimaging.com/docs/products/serial_commands - ASI Quick Start Guide: http://asiimaging.com/docs/command_quick_start + ASI Quick Start Guide: https://asiimaging.com/docs/command_quick_start Note ---- @@ -88,17 +89,25 @@ class ASIStage(StageBase): """ - def __init__(self, microscope_name, device_connection, configuration, device_id=0): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], + device_id: Union[int, Optional] = 0, + ): """Initialize the ASI Stage connection. Parameters ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocessing.managers.DictProxy + configuration : Dict[str, Any] Global configuration of the microscope + device_id : Union[int, Optional] + Device ID for the stage, by default 0 """ super().__init__(microscope_name, device_connection, configuration, device_id) @@ -408,8 +417,9 @@ def scanr(self, start_position_mm, end_position_mm, enc_divide, axis="z"): start_position_mm, end_position_mm, enc_divide, axis ) except MS2000Exception as e: - logger.exception(f"MS2000Exception: {e}") - print(logger.exception()) + error_statement = f"MS2000Exception: {e}" + logger.exception(error_statement) + print(error_statement) return False except KeyError as e: logger.exception(f"ASI Stage - KeyError in scanr: {e}") @@ -428,7 +438,7 @@ def scanv( scan start position end_position_mm: float scan end position - number of lines: int + number_of_lines: int number of steps. overshoot: float overshoot_time ms diff --git a/src/navigate/model/devices/stages/base.py b/src/navigate/model/devices/stages/base.py index c15222582..0a1bd1cf6 100644 --- a/src/navigate/model/devices/stages/base.py +++ b/src/navigate/model/devices/stages/base.py @@ -33,8 +33,8 @@ import logging # from idlelib.debugger_r import DictProxy -from multiprocessing.managers import ListProxy, DictProxy -from typing import Any +from multiprocessing.managers import ListProxy +from typing import Any, Dict, Optional, Union # Third Party Imports @@ -54,8 +54,8 @@ def __init__( self, microscope_name: str, device_connection: Any, - configuration: DictProxy, - device_id: int = 0, + configuration: Dict[str, Any], + device_id: Union[int, Optional] = 0, ) -> None: """Initialize the stage. @@ -65,11 +65,14 @@ def __init__( Name of microscope in configuration device_connection : Any Hardware device to connect to - configuration : DictProxy + configuration : Dict[str, Any] Global configuration of the microscope device_id : int, Optional Device ID. Default is 0. """ + #: Any: Device connection object. + self.device_connection = device_connection + stage_configuration = configuration["configuration"]["microscopes"][ microscope_name ]["stage"] @@ -153,7 +156,7 @@ def get_abs_position(self, axis, axis_abs): """ try: # Get all necessary attributes. - # If we can't we'll move to the error case (e.g., -1e50). + # If we can't, we'll move to the error case (e.g., -1e50). axis_min, axis_max = getattr(self, f"{axis}_min"), getattr( self, f"{axis}_max" ) diff --git a/src/navigate/model/devices/stages/ni.py b/src/navigate/model/devices/stages/ni.py index 4621e524f..920753d7a 100644 --- a/src/navigate/model/devices/stages/ni.py +++ b/src/navigate/model/devices/stages/ni.py @@ -34,6 +34,7 @@ import logging from multiprocessing.managers import ListProxy import time +from typing import Any, Dict, Optional, Union # Third Party Imports import numpy as np @@ -52,25 +53,32 @@ class GalvoNIStage(StageBase): """Galvo Stage Class (only supports one axis) - Generic analog controlled stage. Could be used to control piezos, galvos, - etc. Currently set up to handle National Instruments data acquisition cards. + Generic analog controlled stage. Could be used to control piezoelectric devices, + galvos, etc. Currently set up to handle National Instruments data acquisition cards. Retrieves the volts per micron from the configuration file and uses that to determine the correct voltage to send to the stage. - """ - def __init__(self, microscope_name, device_connection, configuration, device_id=0): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], + device_id: Union[int, Optional] = 0, + ) -> None: """Initialize the Galvo Stage. Parameters ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocessing.managers.DictProxy + configuration : Dict[str, Any] Global configuration of the microscope + device_id : Union[int, Optional] + Device ID of the stage """ super().__init__(microscope_name, device_connection, configuration, device_id) diff --git a/src/navigate/model/devices/stages/pi.py b/src/navigate/model/devices/stages/pi.py index 23fbec471..a07c2db93 100644 --- a/src/navigate/model/devices/stages/pi.py +++ b/src/navigate/model/devices/stages/pi.py @@ -33,6 +33,7 @@ # Standard Imports import logging import time +from typing import Any, Dict, Optional, Union # Third Party Imports from pipython import GCSDevice, pitools, GCSError @@ -88,20 +89,30 @@ def build_PIStage_connection(controller_name, serial_number, stages, reference_m @log_initialization class PIStage(StageBase): - """Physik Instrumente (PI) Stage Class - - Parameters - ---------- - microscope_name : str - Name of microscope in configuration - device_connection : object - Hardware device to connect to - configuration : multiprocessing.managers.DictProxy - Global configuration of the microscope + """Physik Instrumente (PI) Stage Class""" + + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], + device_id: Union[int, Optional] = 0, + ): + """ + Initialize the Physik Instrumente (PI) Stage Class - """ + Parameters + ---------- + microscope_name : str + Name of microscope in configuration + device_connection : Any + Hardware device to connect to + configuration : Dict[str, Any] + Global configuration of the microscope + device_id : int, Optional + Unique identifier for the device, by default 0 + """ - def __init__(self, microscope_name, device_connection, configuration, device_id=0): super().__init__(microscope_name, device_connection, configuration, device_id) # Default mapping from self.axes to corresponding PI axis labelling @@ -128,8 +139,6 @@ def __init__(self, microscope_name, device_connection, configuration, device_id= def __del__(self): """Delete the PI Connection - - Raises ------ GCSError diff --git a/src/navigate/model/devices/stages/sutter.py b/src/navigate/model/devices/stages/sutter.py index 757287b10..4a01a0c62 100644 --- a/src/navigate/model/devices/stages/sutter.py +++ b/src/navigate/model/devices/stages/sutter.py @@ -34,8 +34,7 @@ import time # from idlelib.debugger_r import DictProxy -from multiprocessing.managers import DictProxy -from typing import Any +from typing import Any, Dict # Third-Party Imports from serial import SerialException @@ -86,7 +85,7 @@ def __init__( self, microscope_name: str, device_connection: Any, - configuration: DictProxy, + configuration: Dict[str, Any], device_id: int = 0, ) -> None: """Initialize the SutterStage. @@ -97,7 +96,7 @@ def __init__( Name of the microscope. device_connection : Any MP285 stage connection. - configuration : DictProxy + configuration : Dict[str, Any] Configuration dictionary for the SutterStage. device_id : int, Optional Device ID for the SutterStage. @@ -139,13 +138,17 @@ def __init__( #: str: Resolution of the stage. self.resolution = "low" # "high" - #: int: Speed of the stage. - self.speed = 3000 # 1300 # in units microns/s. + #: int: Speed of the stage in units microns/s. + self.speed = 3000 #: float: Position of the stage along the x-axis. + self.stage_x_pos = 0 + #: float: Position of the stage along the y-axis. + self.stage_y_pos = 0 + #: float: Position of the stage along the z-axis. - self.stage_x_pos, self.stage_y_pos, self.stage_z_pos = 0, 0, 0 + self.stage_z_pos = 0 # Set the resolution and velocity of the stage try: diff --git a/src/navigate/model/devices/zoom/synthetic.py b/src/navigate/model/devices/zoom/synthetic.py index 2ee9f2972..c2cff2ffa 100644 --- a/src/navigate/model/devices/zoom/synthetic.py +++ b/src/navigate/model/devices/zoom/synthetic.py @@ -29,10 +29,10 @@ # IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE # POSSIBILITY OF SUCH DAMAGE. -# # Standard Library Imports import logging +from typing import Any, Dict # Third Party Imports @@ -49,20 +49,25 @@ class SyntheticZoom(ZoomBase): """SyntheticZoom Class - Controls the SyntheticZoom Servo.""" - def __init__(self, microscope_name, device_connection, configuration): + def __init__( + self, + microscope_name: str, + device_connection: Any, + configuration: Dict[str, Any], + ) -> None: """Initialize the SyntheticZoom Servo. Parameters ---------- microscope_name : str Name of microscope in configuration - device_connection : object + device_connection : Any Hardware device to connect to - configuration : multiprocessing.managers.DictProxy + configuration : Dict[str, Any] Global configuration of the microscope """ super().__init__(microscope_name, device_connection, configuration) - def __del__(self): + def __del__(self) -> None: """Delete the SyntheticZoom Servo.""" pass diff --git a/src/navigate/model/metadata_sources/metadata.py b/src/navigate/model/metadata_sources/metadata.py index d6730c352..77deeacd9 100644 --- a/src/navigate/model/metadata_sources/metadata.py +++ b/src/navigate/model/metadata_sources/metadata.py @@ -32,13 +32,15 @@ # Standard Imports import os import logging -from typing import Optional +from typing import Optional, Dict, Any +from multiprocessing.managers import DictProxy + +# Third-party Imports # Local Imports from navigate.tools import xml_tools from navigate import __version__, __commit__ -from multiprocessing.managers import DictProxy # Logger Setup p = __name__.split(".")[1] @@ -107,12 +109,12 @@ def configuration(self) -> Optional[DictProxy]: return self._configuration @configuration.setter - def configuration(self, configuration: DictProxy) -> None: + def configuration(self, configuration: Dict[str, Any]) -> None: """Set configuration dictionary Parameters ---------- - configuration : DictProxy + configuration : Dict[str, Any] Configuration dictionary """ self._configuration = configuration diff --git a/src/navigate/model/microscope.py b/src/navigate/model/microscope.py index 92ba5d732..3ed1eea53 100644 --- a/src/navigate/model/microscope.py +++ b/src/navigate/model/microscope.py @@ -32,8 +32,9 @@ # Standard Library imports import logging import importlib # noqa: F401 -from multiprocessing.managers import ListProxy, DictProxy +from multiprocessing.managers import ListProxy import reprlib +from typing import Any, Dict # Third-party imports @@ -52,7 +53,7 @@ class Microscope: def __init__( self, name: str, - configuration: DictProxy, + configuration: Dict[str, Any], devices_dict: dict, is_synthetic=False, is_virtual=False, @@ -63,7 +64,7 @@ def __init__( ---------- name : str Name of the microscope. - configuration : DictProxy + configuration : Dict[str, Any] Configuration dictionary. devices_dict : dict Dictionary of devices. diff --git a/src/navigate/model/waveforms.py b/src/navigate/model/waveforms.py index f4afe5ece..e6a0ec831 100644 --- a/src/navigate/model/waveforms.py +++ b/src/navigate/model/waveforms.py @@ -469,18 +469,13 @@ def smooth_waveform(waveform, percent_smoothing=10): ---------- waveform : np.array The waveform to be smoothed - percent_smoothing : int + percent_smoothing : float The percentage of the waveform to be smoothed Returns ------- smoothed_waveform : np.array The smoothed waveform - - Examples - -------- - >>> smoothed_waveform = smooth_waveform(waveform, percent_smoothing) - """ waveform_length = np.size(waveform) window_length = int(np.ceil(waveform_length * percent_smoothing / 100)) From 6589ca7eef749759b55aefdc1984f638ea545d9c Mon Sep 17 00:00:00 2001 From: Kevin Dean <42547789+AdvancedImagingUTSW@users.noreply.github.com> Date: Tue, 8 Oct 2024 13:08:17 -0500 Subject: [PATCH 5/5] Misuse of Optional Learning the hardway... --- .../model/device_startup_functions.py | 104 +++++++++--------- .../model/devices/APIs/sutter/MP285.py | 12 +- src/navigate/model/devices/stages/asi.py | 6 +- .../model/devices/stages/asi_MSTwoThousand.py | 11 +- src/navigate/model/devices/stages/base.py | 6 +- src/navigate/model/devices/stages/ni.py | 6 +- src/navigate/model/devices/stages/pi.py | 6 +- src/navigate/model/devices/stages/sutter.py | 2 +- .../model/metadata_sources/bdv_metadata.py | 2 +- .../model/metadata_sources/metadata.py | 6 +- .../metadata_sources/ome_tiff_metadata.py | 2 +- src/navigate/plugins/plugin_manager.py | 10 +- 12 files changed, 89 insertions(+), 84 deletions(-) diff --git a/src/navigate/model/device_startup_functions.py b/src/navigate/model/device_startup_functions.py index e100c0973..218177c98 100644 --- a/src/navigate/model/device_startup_functions.py +++ b/src/navigate/model/device_startup_functions.py @@ -35,7 +35,7 @@ import time import importlib from multiprocessing.managers import ListProxy -from typing import Callable, Tuple, Any, Type, Dict, Union, Optional +from typing import Callable, Tuple, Any, Type, Dict, Optional # Third Party Imports @@ -84,9 +84,9 @@ def auto_redial( The function or class (`__init__()` method) that connects to a device. args : Tuple[Any, ...] Positional arguments to pass to the `func`. - n_tries : int, Optional + n_tries : int The number of attempts to retry the connection. Default is 10. - exception : Type[Exception], Optional + exception : Type[Exception] The exception type to catch and handle during connection attempts. Default is `Exception`. **kwargs : Any @@ -158,7 +158,7 @@ def build_connection( Function that builds the connection to the device. args : Tuple Arguments to the build_connection_function - exception : Exception, Optional + exception : Type[Exception] Exception to catch when building the connection Returns @@ -193,9 +193,9 @@ def load_camera_connection( ---------- configuration : Dict[str, Any] Global configuration of the microscope - camera_id : int, Optional + camera_id : int Device ID (0, 1...) - is_synthetic: bool, Optional + is_synthetic: bool Whether it is a synthetic hardware Returns @@ -250,7 +250,7 @@ def start_camera( device_connection: Any, configuration: Dict[str, Any], is_synthetic: bool = False, - plugin_devices=None, + plugin_devices: dict = None, ) -> CameraBase: """Initialize the camera class. @@ -262,9 +262,9 @@ def start_camera( Hardware device to connect to configuration : Dict[str, Any] Global configuration of the microscope - is_synthetic : bool, Optional + is_synthetic : bool Run synthetic version of hardware. Default is False. - plugin_devices : dict, Optional + plugin_devices : dict Dictionary of plugin devices. Default is None. Returns @@ -339,7 +339,7 @@ def load_mirror(configuration: Dict[str, Any], is_synthetic: bool = False) -> An ---------- configuration : Dict[str, Any] Global configuration of the microscope - is_synthetic : bool, Optional + is_synthetic : bool Run synthetic version of hardware. Default is False. Returns @@ -369,7 +369,7 @@ def start_mirror( device_connection: Any, configuration: Dict[str, Any], is_synthetic: bool = False, - plugin_devices: Union[Dict, None, Optional] = None, + plugin_devices: Optional[Dict] = None, ) -> MirrorBase: """Initialize the mirror class. @@ -381,9 +381,9 @@ def start_mirror( Hardware device to connect to configuration : Dict[str, Any] Global configuration of the microscope - is_synthetic : bool, Optional + is_synthetic : bool Run synthetic version of hardware. Default is False. - plugin_devices : dict, Optional + plugin_devices : Optional[Dict] Dictionary of plugin devices. Default is None. Returns @@ -421,7 +421,9 @@ def start_mirror( def load_stages( - configuration: Dict[str, Any], is_synthetic: bool = False, plugin_devices=None + configuration: Dict[str, Any], + is_synthetic: bool = False, + plugin_devices: Optional[Dict] = None, ) -> Any: """Initialize the stage API. @@ -432,9 +434,9 @@ def load_stages( ---------- configuration : Dict[str, Any] Global configuration of the microscope - is_synthetic : bool, Optional + is_synthetic : bool Run synthetic version of hardware. Default is False. - plugin_devices : dict, Optional + plugin_devices : Optional[Dict] Dictionary of plugin devices. Default is None. Returns @@ -646,7 +648,7 @@ def start_stage( configuration: Dict[str, Any], id: int = 0, is_synthetic: bool = False, - plugin_devices=None, + plugin_devices: Optional[Dict] = None, ) -> StageBase: """Initialize the Stage class. @@ -661,11 +663,11 @@ def start_stage( Hardware device to connect to configuration : Dict[str, Any] Global configuration of the microscope - id : int, Optional + id : int ID of the stage. Default is 0. - is_synthetic : bool, Optional + is_synthetic : bool Run synthetic version of hardware. Default is False. - plugin_devices : dict, Optional + plugin_devices : Optional[Dict] Dictionary of plugin devices. Default is None. Returns @@ -759,7 +761,9 @@ def start_stage( def load_zoom_connection( - configuration: Dict[str, Any], is_synthetic: bool = False, plugin_devices=None + configuration: Dict[str, Any], + is_synthetic: bool = False, + plugin_devices: Optional[Dict] = None, ) -> Any: """Initializes the Zoom class on a dedicated thread. @@ -772,12 +776,12 @@ def load_zoom_connection( Global configuration of the microscope is_synthetic : bool Run synthetic version of hardware? - plugin_devices : dict, Optional + plugin_devices : Optional[Dict] Dictionary of plugin devices. Default is None. Returns ------- - Zoom : class + Zoom : Any Zoom class. """ @@ -820,7 +824,7 @@ def start_zoom( device_connection: Any, configuration: Dict[str, Any], is_synthetic: bool = False, - plugin_devices=None, + plugin_devices: Optional[dict] = None, ) -> ZoomBase: """Initializes the zoom class on a dedicated thread. @@ -834,9 +838,9 @@ def start_zoom( Hardware device to connect to configuration : Dict[str, Any] Global configuration of the microscope - is_synthetic : bool, Optional + is_synthetic : bool Run synthetic version of hardware. Default is False. - plugin_devices : dict, Optional + plugin_devices : Optional[dict] Dictionary of plugin devices. Default is None. Returns @@ -895,7 +899,9 @@ def start_zoom( def load_filter_wheel_connection( - device_info: Dict[str, Any], is_synthetic=False, plugin_devices=None + device_info: Dict[str, Any], + is_synthetic: bool = False, + plugin_devices: Optional[dict] = None, ) -> Any: """Initializes the Filter Wheel class on a dedicated thread. @@ -906,9 +912,9 @@ def load_filter_wheel_connection( ---------- device_info : Dict[str, Any] filter wheel device configuration - is_synthetic : bool, Optional + is_synthetic : bool Run synthetic version of hardware, Default is False. - plugin_devices : dict, Optional + plugin_devices : Optional[dict] Dictionary of plugin devices. Default is None. Returns @@ -987,7 +993,7 @@ def start_filter_wheel( configuration: Dict[str, Any], id: int = 0, is_synthetic: bool = False, - plugin_devices=None, + plugin_devices: Optional[dict] = None, ) -> FilterWheelBase: """Initialize the filter wheel class. @@ -1001,11 +1007,11 @@ def start_filter_wheel( Device connection configuration : Dict[str, Any] Global configuration of the microscope - id : int, Optional + id : int Index of filter_wheel in the configuration dictionary. Default is 0. - is_synthetic : bool, Optional + is_synthetic : bool Run synthetic version of hardware. Default is False. - plugin_devices : dict, Optional + plugin_devices : Optional[dict] Dictionary of plugin devices. Default is None. Returns @@ -1091,7 +1097,7 @@ def start_daq(configuration: Dict[str, Any], is_synthetic: bool = False) -> DAQB ---------- configuration : Dict[str, Any] Global configuration of the microscope - is_synthetic : bool, Optional + is_synthetic : bool Run synthetic version of hardware. Default is False. Returns @@ -1124,11 +1130,11 @@ def start_shutter( device_connection: Any, configuration: Dict[str, Any], is_synthetic: bool = False, - plugin_devices: dict = None, + plugin_devices: Optional[dict] = None, ) -> ShutterBase: """Initializes the shutter class on a dedicated thread. - Initializes the shutters: ThorlabsShutter or SyntheticShutter + Initializes the shutters: Thorlabs, Shutter or SyntheticShutter Shutters are triggered via digital outputs on the NI DAQ Card Thus, requires both to be enabled. @@ -1140,9 +1146,9 @@ def start_shutter( Hardware device to connect to configuration : Dict[str, Any] Global configuration of the microscope - is_synthetic : bool, Optional + is_synthetic : bool Run synthetic version of hardware. Default is False. - plugin_devices : dict, Optional + plugin_devices : Optional[dict] Dictionary of plugin devices. Default is None. Returns @@ -1201,7 +1207,7 @@ def start_lasers( configuration: Dict[str, Any], id: int = 0, is_synthetic: bool = False, - plugin_devices: dict = None, + plugin_devices: Optional[dict] = None, ) -> LaserBase: """Initializes the lasers. @@ -1216,11 +1222,11 @@ def start_lasers( Hardware device to connect to configuration : Dict[str, Any] Global configuration of the microscope - id : int, Optional + id : int Index of laser in laser list in configuration dictionary. Default is 0. - is_synthetic : bool, Optional + is_synthetic : bool Run synthetic version of hardware. Default is False. - plugin_devices : dict, Optional + plugin_devices : Optional[dict] Dictionary of plugin devices. Default is None. Returns @@ -1278,7 +1284,7 @@ def start_remote_focus_device( device_connection: Any, configuration: Dict[str, Any], is_synthetic: bool = False, - plugin_devices=None, + plugin_devices: Optional[dict] = None, ) -> RemoteFocusBase: """Initializes the remote focus class. @@ -1293,9 +1299,9 @@ def start_remote_focus_device( Hardware device to connect to configuration : Dict[str, Any] Global configuration of the microscope - is_synthetic : bool, Optional + is_synthetic : bool Run synthetic version of hardware. Default is False. - plugin_devices : dict, Optional + plugin_devices : Optional[dict] Dictionary of plugin devices. Default is None. Returns @@ -1363,7 +1369,7 @@ def start_galvo( configuration: Dict[str, Any], id: int = 0, is_synthetic: bool = False, - plugin_devices=None, + plugin_devices: Optional[dict] = None, ) -> GalvoBase: """Initializes the Galvo class. @@ -1377,11 +1383,11 @@ def start_galvo( Hardware device to connect to configuration : Dict[str, Any] Global configuration of the microscope - id : int, Optional + id : int Index of galvo in the configuration dictionary. Default is 0. - is_synthetic : bool, Optional + is_synthetic : bool Run synthetic version of hardware. Default is False. - plugin_devices : dict, Optional + plugin_devices : Optional[dict] Dictionary of plugin devices. Default is None. Returns diff --git a/src/navigate/model/devices/APIs/sutter/MP285.py b/src/navigate/model/devices/APIs/sutter/MP285.py index aead217c1..a677067b3 100644 --- a/src/navigate/model/devices/APIs/sutter/MP285.py +++ b/src/navigate/model/devices/APIs/sutter/MP285.py @@ -304,12 +304,12 @@ def get_current_position( Returns ------- - x_pos : float - X position in microns - y_pos : float - Y position in microns - z_pos : float - Z position in microns + position : Tuple[Optional[float], Optional[float], Optional[float]] + The x, y, and z positions in microns if available. + + - x_pos is the X position in microns. None if the position is not available. + - y_pos is the Y position in microns. None if the position is not available. + - z_pos is the Z position in microns. None if the position is not available. """ # print("calling get_current_position") diff --git a/src/navigate/model/devices/stages/asi.py b/src/navigate/model/devices/stages/asi.py index c66f7a8ac..906a99545 100644 --- a/src/navigate/model/devices/stages/asi.py +++ b/src/navigate/model/devices/stages/asi.py @@ -32,7 +32,7 @@ # Standard Imports import logging import time -from typing import Any, Dict, Optional, Union +from typing import Any, Dict # Third Party Imports @@ -99,7 +99,7 @@ def __init__( microscope_name: str, device_connection: Any, configuration: Dict[str, Any], - device_id: Union[int, Optional] = 0, + device_id: int = 0, ): """Initialize the ASI Stage connection. @@ -111,7 +111,7 @@ def __init__( Hardware device to connect to configuration : Dict[str, Any] Global configuration of the microscope - device_id : Union[int, Optional] + device_id : int Device ID for the stage, defaults to 0 """ super().__init__(microscope_name, device_connection, configuration, device_id) diff --git a/src/navigate/model/devices/stages/asi_MSTwoThousand.py b/src/navigate/model/devices/stages/asi_MSTwoThousand.py index 83bf3c1df..269d07388 100644 --- a/src/navigate/model/devices/stages/asi_MSTwoThousand.py +++ b/src/navigate/model/devices/stages/asi_MSTwoThousand.py @@ -32,7 +32,7 @@ # Standard Imports import logging import time -from typing import Any, Dict, Optional, Union +from typing import Any, Dict # Third Party Imports @@ -86,7 +86,6 @@ class ASIStage(StageBase): Note ---- ASI firmware requires all distances to be in a 10th of a micron. - """ def __init__( @@ -94,7 +93,7 @@ def __init__( microscope_name: str, device_connection: Any, configuration: Dict[str, Any], - device_id: Union[int, Optional] = 0, + device_id: int = 0, ): """Initialize the ASI Stage connection. @@ -106,7 +105,7 @@ def __init__( Hardware device to connect to configuration : Dict[str, Any] Global configuration of the microscope - device_id : Union[int, Optional] + device_id : int Device ID for the stage, by default 0 """ super().__init__(microscope_name, device_connection, configuration, device_id) @@ -475,7 +474,7 @@ def move_axis_relative(self, axis, distance, wait_until_done=False): distance : float The distance to move relative to the current position, in micrometers for XYZ axes. - wait_until_done : bool, optional + wait_until_done : bool Whether to wait until the stage has moved to its new position, by default False. @@ -525,7 +524,7 @@ def scan_axis_triggered_move( The desired end position of the stage along the specified axis. axis : str The axis along which the stage will be moved (e.g., 'x', 'y', 'z'). - ttl_triggered : bool, optional + ttl_triggered : bool Whether to trigger the move using TTL signal, by default False. Returns diff --git a/src/navigate/model/devices/stages/base.py b/src/navigate/model/devices/stages/base.py index 0a1bd1cf6..cab5c204f 100644 --- a/src/navigate/model/devices/stages/base.py +++ b/src/navigate/model/devices/stages/base.py @@ -34,7 +34,7 @@ # from idlelib.debugger_r import DictProxy from multiprocessing.managers import ListProxy -from typing import Any, Dict, Optional, Union +from typing import Any, Dict # Third Party Imports @@ -55,7 +55,7 @@ def __init__( microscope_name: str, device_connection: Any, configuration: Dict[str, Any], - device_id: Union[int, Optional] = 0, + device_id: int = 0, ) -> None: """Initialize the stage. @@ -67,7 +67,7 @@ def __init__( Hardware device to connect to configuration : Dict[str, Any] Global configuration of the microscope - device_id : int, Optional + device_id : int Device ID. Default is 0. """ #: Any: Device connection object. diff --git a/src/navigate/model/devices/stages/ni.py b/src/navigate/model/devices/stages/ni.py index 920753d7a..5db268567 100644 --- a/src/navigate/model/devices/stages/ni.py +++ b/src/navigate/model/devices/stages/ni.py @@ -34,7 +34,7 @@ import logging from multiprocessing.managers import ListProxy import time -from typing import Any, Dict, Optional, Union +from typing import Any, Dict # Third Party Imports import numpy as np @@ -65,7 +65,7 @@ def __init__( microscope_name: str, device_connection: Any, configuration: Dict[str, Any], - device_id: Union[int, Optional] = 0, + device_id: int = 0, ) -> None: """Initialize the Galvo Stage. @@ -77,7 +77,7 @@ def __init__( Hardware device to connect to configuration : Dict[str, Any] Global configuration of the microscope - device_id : Union[int, Optional] + device_id : int Device ID of the stage """ super().__init__(microscope_name, device_connection, configuration, device_id) diff --git a/src/navigate/model/devices/stages/pi.py b/src/navigate/model/devices/stages/pi.py index a07c2db93..cf8a129c1 100644 --- a/src/navigate/model/devices/stages/pi.py +++ b/src/navigate/model/devices/stages/pi.py @@ -33,7 +33,7 @@ # Standard Imports import logging import time -from typing import Any, Dict, Optional, Union +from typing import Any, Dict # Third Party Imports from pipython import GCSDevice, pitools, GCSError @@ -96,7 +96,7 @@ def __init__( microscope_name: str, device_connection: Any, configuration: Dict[str, Any], - device_id: Union[int, Optional] = 0, + device_id: int = 0, ): """ Initialize the Physik Instrumente (PI) Stage Class @@ -109,7 +109,7 @@ def __init__( Hardware device to connect to configuration : Dict[str, Any] Global configuration of the microscope - device_id : int, Optional + device_id : int Unique identifier for the device, by default 0 """ diff --git a/src/navigate/model/devices/stages/sutter.py b/src/navigate/model/devices/stages/sutter.py index 4a01a0c62..97510734b 100644 --- a/src/navigate/model/devices/stages/sutter.py +++ b/src/navigate/model/devices/stages/sutter.py @@ -98,7 +98,7 @@ def __init__( MP285 stage connection. configuration : Dict[str, Any] Configuration dictionary for the SutterStage. - device_id : int, Optional + device_id : int Device ID for the SutterStage. Raises diff --git a/src/navigate/model/metadata_sources/bdv_metadata.py b/src/navigate/model/metadata_sources/bdv_metadata.py index 356729fc9..a06ece6e9 100644 --- a/src/navigate/model/metadata_sources/bdv_metadata.py +++ b/src/navigate/model/metadata_sources/bdv_metadata.py @@ -334,7 +334,7 @@ def stage_positions_to_affine_matrix( The z position of the stage. theta : float The theta position of the stage. - f : Optional[float], optional + f : Optional[float] The focus position of the stage, by default None Returns diff --git a/src/navigate/model/metadata_sources/metadata.py b/src/navigate/model/metadata_sources/metadata.py index 77deeacd9..d06711203 100644 --- a/src/navigate/model/metadata_sources/metadata.py +++ b/src/navigate/model/metadata_sources/metadata.py @@ -103,7 +103,7 @@ def configuration(self) -> Optional[DictProxy]: Returns ------- - Optional[DictProxy] + configuration : Optional[DictProxy] Configuration dictionary """ return self._configuration @@ -289,7 +289,7 @@ def write_xml( File name file_type : str File type - root : Optional[str], optional + root : Optional[str] Root, by default None """ xml = '\n' # XML file header @@ -313,7 +313,7 @@ def to_xml(self, file_type: str, root: Optional[str] = None, **kw) -> str: ---------- file_type : str File type - root : Optional[str], optional + root : Optional[str] Root, by default None **kw Keyword arguments diff --git a/src/navigate/model/metadata_sources/ome_tiff_metadata.py b/src/navigate/model/metadata_sources/ome_tiff_metadata.py index 85ee82237..6393d62e1 100644 --- a/src/navigate/model/metadata_sources/ome_tiff_metadata.py +++ b/src/navigate/model/metadata_sources/ome_tiff_metadata.py @@ -66,7 +66,7 @@ def ome_tiff_xml_dict( Time point index, by default 0 file_name : Union[str, list, None], optional File name or list of file names, by default None - uid : Union[str, list, None], optional + uid : Union[str, list, None] Unique identifier or list of unique identifiers, by default None views : Optional[list], optional List of views, by default None diff --git a/src/navigate/plugins/plugin_manager.py b/src/navigate/plugins/plugin_manager.py index d115cf162..6e64f8f01 100644 --- a/src/navigate/plugins/plugin_manager.py +++ b/src/navigate/plugins/plugin_manager.py @@ -38,7 +38,7 @@ import os import inspect import logging -from typing import Optional +from typing import Optional, Any # Local application imports from navigate.tools.file_functions import load_yaml_file @@ -108,7 +108,7 @@ def load_config(package_name: str) -> dict: return plugin_config @staticmethod - def load_controller(package_name: str, controller_name: str) -> Optional[type]: + def load_controller(package_name: str, controller_name: str) -> Optional[Any]: """Load controller Parameters @@ -120,7 +120,7 @@ def load_controller(package_name: str, controller_name: str) -> Optional[type]: Returns ------- - controller_class : type or None + controller_class : Optional[Any] The dynamically loaded controller class, or `None` if the controller module or class cannot be found. The return type can be: @@ -139,7 +139,7 @@ def load_controller(package_name: str, controller_name: str) -> Optional[type]: return None @staticmethod - def load_view(package_name: str, frame_name: str) -> Optional[type]: + def load_view(package_name: str, frame_name: str) -> Optional[Any]: """Load view Parameters @@ -151,7 +151,7 @@ def load_view(package_name: str, frame_name: str) -> Optional[type]: Returns ------- - view_class : type or None + view_class : Optional[Any] The dynamically loaded view class, or `None` if the view module or class cannot be found. The return type can be: