Merge pull request #958 from TheDeanLab/pr956-update-from-develop

Pr956 update from develop

Author: AdvancedImagingUTSW

src/navigate/config/config.py

@@ -341,6 +341,7 @@ def verify_experiment_config(manager, configuration):
         "celltype": "MV3",
         "label": "GFP",
         "file_type": "TIFF",
+        "prefix": "Cell_",
         "date": time.strftime("%Y-%m-%d"),
         "solvent": "BABB",
     }

src/navigate/model/devices/APIs/asi/asi_MS2000_controller.py

@@ -94,6 +94,7 @@ def __init__(self, com_port: str, baud_rate: int, verbose: bool = False):
         #: float: Last time a command was sent to the MS2000 Controller
         self._last_cmd_send_time = time.perf_counter()
 
+
     @staticmethod
     def scan_ports() -> list[str]:
         """Scans for available COM ports
@@ -109,6 +110,7 @@ def scan_ports() -> list[str]:
         com_ports.sort(key=lambda value: int(value[3:]))
         return com_ports
 
+
     def connect_to_serial(
         self,
         rx_size: int = 12800,
@@ -143,8 +145,6 @@ def connect_to_serial(
 
         # set the size of the rx and tx buffers before calling open
         self.serial_port.set_buffer_size(rx_size, tx_size)
-
-        # try to open the serial port
         try:
             self.serial_port.open()
         except SerialException:
@@ -164,7 +164,8 @@ def connect_to_serial(
             )
 
             #: list[str]: Default axes sequence of the MS2000 Controller
-            self.default_axes_sequence = self.get_default_motor_axis_sequence()
+            self.default_axes_sequence = ["X", "Y", "Z"] # self.get_default_motor_axis_sequence()
+
 
     def get_default_motor_axis_sequence(self) -> None:
         """Get the default motor axis sequence from the ASI device
@@ -187,6 +188,7 @@ def get_default_motor_axis_sequence(self) -> None:
 
         return default_axes_sequence
 
+
     def set_feedback_alignment(self, axis, aa):
         """Set the stage feedback alignment.
 
@@ -200,6 +202,7 @@ def set_feedback_alignment(self, axis, aa):
         self.send_command(f"AZ {axis}\r")
         self.read_response()
 
+
     def set_backlash(self, axis, val):
         """Enable/disable stage backlash correction.
 
@@ -210,10 +213,10 @@ def set_backlash(self, axis, val):
         val : float
             Distance of anti-backlash motion [mm]
         """
-        cmd_str = f"B {axis}={val:.7f}\r"
         self.send_command(f"B {axis}={val:.7f}\r")
         self.read_response()
 
+
     def set_finishing_accuracy(self, axis, ac):
         """Set the stage finishing accuracy.
 
@@ -227,6 +230,7 @@ def set_finishing_accuracy(self, axis, ac):
         self.send_command(f"PC {axis}={ac:.7f}\r")
         self.read_response()
 
+
     def set_error(self, axis, ac):
         """
         Set the stage drift error
@@ -240,14 +244,15 @@ def set_error(self, axis, ac):
         """
         self.send_command(f"E {axis}={ac:.7f}\r")
         self.read_response()
-        
+       
 
     def disconnect_from_serial(self) -> None:
         """Disconnect from the serial port if it's open."""
         if self.is_open():
             self.serial_port.close()
             self.report_to_console("Disconnected from the serial port.")
 
+
     def is_open(self) -> bool:
         """Returns True if the serial port exists and is open.
 
@@ -259,6 +264,7 @@ def is_open(self) -> bool:
         # short circuits if serial port is None
         return self.serial_port and self.serial_port.is_open
 
+
     def report_to_console(self, message: str) -> None:
         """Print message to the output device, usually the console.
 
@@ -268,10 +274,10 @@ def report_to_console(self, message: str) -> None:
             Message to print to the output device
         """
         # useful if we want to output data to something other than the console
-        # (ui element etc)
         if self.verbose:
             print(message)
 
+            
     def wait_for_device(self, report: bool = False) :
         """Waits for the all motors to stop moving."""
         if not report:
@@ -283,6 +289,7 @@ def wait_for_device(self, report: bool = False) :
             busy = self.is_device_busy()
         self.report = temp
 
+        
     def send_command(self, cmd: bytes) -> None:
         """Send a serial command to the device.
 
@@ -309,6 +316,7 @@ def send_command(self, cmd: bytes) -> None:
 
         # sleep to avoid error, emperically found this made it work
         time.sleep(0.1)
+       
 
     def read_response(self) -> str:
         """Read a line from the serial response.
@@ -320,16 +328,15 @@ def read_response(self) -> str:
         """
         response = self.serial_port.readline()
         self.safe_to_write.set()
-
         response = response.decode(encoding="ascii")
         # Remove leading and trailing empty spaces
         self.report_to_console(f"Received Response: {response.strip()}")
-        
         if response.startswith(":N"):
             raise MS2000Exception(response)
 
         return response  # in case we want to read the response
 
+
     def moverel(self, x: int = 0, y: int = 0, z: int = 0) -> None:
         """Move the stage with a relative move on multiple axes.
 
@@ -345,6 +352,7 @@ def moverel(self, x: int = 0, y: int = 0, z: int = 0) -> None:
         self.send_command(f"MOVREL X={x} Y={y} Z={z}\r")
         self.read_response()
 
+
     def moverel_axis(self, axis: str, distance: float) -> None:
         """Move the stage with a relative move on one axis
 
@@ -358,6 +366,7 @@ def moverel_axis(self, axis: str, distance: float) -> None:
         self.send_command(f"MOVREL {axis}={round(distance, 6)}\r")
         self.read_response()
 
+
     def move(self, pos_dict) -> None:
         """Move the stage with an absolute move on multiple axes
 
@@ -372,6 +381,7 @@ def move(self, pos_dict) -> None:
         self.send_command(f"MOVE {pos_str}\r")
         self.read_response()
 
+        
     def move_axis(self, axis: str, distance: float) -> None:
         """Move the stage with an absolute move on one axis
 
@@ -385,6 +395,7 @@ def move_axis(self, axis: str, distance: float) -> None:
         self.send_command(f"MOVE {axis}={round(distance, 6)}\r")
         self.read_response()
 
+
     def set_max_speed(self, axis: str, speed: float) -> None:
         """Set the speed on a specific axis. Speed is in mm/s.
 
@@ -401,6 +412,7 @@ def set_max_speed(self, axis: str, speed: float) -> None:
         self.send_command(f"SPEED {axis}={speed}\r")
         self.read_response()
 
+
     def get_axis_position(self, axis: str) -> int:
         """Return the position of the stage in ASI units (tenths of microns).
 
@@ -415,12 +427,10 @@ def get_axis_position(self, axis: str) -> int:
         """
         self.send_command(f"WHERE {axis}\r")
         response = self.read_response()
-        # try:
         pos = float(response.split(" ")[1])
-        # except:
-        #     pos = float('Inf')
         return pos
 
+
     def get_axis_position_um(self, axis: str) -> float:
         """Return the position of the stage in microns.
 
@@ -437,6 +447,7 @@ def get_axis_position_um(self, axis: str) -> float:
         response = self.read_response()
         return float(response.split(" ")[1]) / 10.0
 
+
     def get_position(self, axes) -> dict:
         """Return current stage position in ASI units.
 
@@ -465,8 +476,7 @@ def get_position(self, axes) -> dict:
             cmd = f"WHERE {' '.join(axes)}\r"
             self.send_command(cmd)
             response = self.read_response()
-
-            # return response.split(" ")[1:-1]
+            
             pos = response.split(" ")
             axes_seq = list(
                 filter(
@@ -483,14 +493,17 @@ def get_position(self, axes) -> dict:
                     # Report position failed. Don't crash, we can try again.
                     pass
             return axis_dict
+        
         else:
             result = {}
             for axis in axes:
                 result[axis] = self.get_axis_position(axis)
             return result
 
+
     # Utility Functions
 
+
     def is_axis_busy(self, axis: str) -> bool:
         """Returns True if the axis is busy.
 
@@ -507,6 +520,7 @@ def is_axis_busy(self, axis: str) -> bool:
         res = self.read_response()
         return "B" in res
 
+
     def is_device_busy(self) -> bool:
         """Returns True if any axis is busy.
 
@@ -519,17 +533,20 @@ def is_device_busy(self) -> bool:
         res = self.read_response()
         return "B" in res
 
+
     def wait_for_device(self, timeout: float = 1.75) -> None:
         """Waits for the all motors to stop moving.
 
         timeout : float
             Timeout in seconds. Default is 1.75 seconds.
         """
+        
         if self.verbose:
             print("Waiting for device...")
         busy = self.is_device_busy()
         waiting_time = 0.0
-
+        
+        # t_start = time.time()      
         while busy:
             waiting_time += 0.001
             if waiting_time >= timeout:
@@ -540,6 +557,7 @@ def wait_for_device(self, timeout: float = 1.75) -> None:
         if self.verbose:
             print(f"Waited {waiting_time:.2f} s")
 
+
     def stop(self):
         """Stop all stage movement immediately"""
 
@@ -548,6 +566,7 @@ def stop(self):
         if self.verbose:
             print("ASI Stages stopped successfully")
 
+
     def set_speed(self, speed_dict):
         """Set speed
 
@@ -560,6 +579,7 @@ def set_speed(self, speed_dict):
         self.send_command(f"S {axes}")
         self.read_response()
 
+
     def set_speed_as_percent_max(self, pct):
         """Set speed as a percentage of the maximum speed
 
@@ -601,6 +621,7 @@ def get_speed(self, axis: str):
         response = self.read_response()
         return float(response.split("=")[1])
 
+
     def get_encoder_counts_per_mm(self, axis: str):
         """Get encoder counts pre mm of axis
 
@@ -619,6 +640,7 @@ def get_encoder_counts_per_mm(self, axis: str):
         response = self.read_response()
         return float(response.split("=")[1].split()[0])
 
+
     def scanr(
         self,
         start_position_mm: float,
@@ -655,6 +677,7 @@ def scanr(
         self.send_command(command)
         self.read_response()
 
+
     def scanv(
         self,
         start_position_mm: float,
@@ -689,6 +712,7 @@ def scanv(
         self.send_command(command)
         self.read_response()
 
+
     def start_scan(self, axis: str, is_single_axis_scan: bool = True):
         """
         Start scan
@@ -703,11 +727,13 @@ def start_scan(self, axis: str, is_single_axis_scan: bool = True):
         self.send_command("SCAN")
         self.read_response()
 
+
     def stop_scan(self):
         """Stop scan."""
         self.send_command("SCAN P")
         self.read_response()
 
+
     def is_moving(self):
         """Check to see if the stage is moving.
 
@@ -744,6 +770,7 @@ def is_moving(self):
             print("WARNING: WAIT UNTIL DONE RECEIVED NO RESPONSE")
             return False
 
+
     def set_triggered_move(self, axis):
         '''
         Set ASI to repeat the latest relative movement  

src/navigate/model/devices/camera/photometrics.py

@@ -214,9 +214,6 @@ def set_readout_direction(self, mode):
             Scan direction options: {'Down': 0, 'Up': 1, 'Down/Up Alternate': 2}
 
         """
-        # print("available scan directions on camera are:")
-        # print(str(self.camera_controller.prog_scan_dirs))
-
         if mode == "Top-to-Bottom":
             #  'Down' readout direction
             self.camera_controller.prog_scan_dir = 0
@@ -359,7 +356,6 @@ def set_binning(self, binning_string):
         }
         if binning_string not in binning_dict.keys():
             logger.debug(f"can't set binning to {binning_string}")
-            print(f"Can't set binning to {binning_string}")
             return False
         self.camera_controller.binning = binning_dict[binning_string]
         idx = binning_string.index("x")
@@ -442,15 +438,11 @@ def initialize_image_series(self, data_buffer=None, number_of_frames=100):
             self.camera_controller.exp_mode = "Edge Trigger"
             self.camera_controller.prog_scan_line_delay = self._scandelay
             self.camera_controller.exp_out_mode = 4
-            print(
-                "camera ready to acquire programmable scan mode "
-                "with scandelay {}".format(self._scandelay)
-            )
+
         else:
             # Normal mode
             self.camera_controller.exp_out_mode = "Any Row"
             self.camera_controller.exp_mode = "Edge Trigger"
-            print("camera ready to acquire static light sheet mode")
 
         # Prepare for buffered acquisition
         #: int: Number of frames
@@ -496,10 +488,14 @@ def _receive_images(self):
             # Delete copied frame for memory management
             frame = None
             del frame
+
+            frame_to_return = [self._frames_received]
             self._frames_received += 1
+            # check to make sure the next frame exist in buffer
             if self._frames_received >= self._numberofframes:
                 self._frames_received = 0
-            return [self._frames_received - 1]
+            return frame_to_return
+
         except Exception as e:
             print(str(e))