czbiohub-sf · jeskesen · Aug 12, 2025 · Jun 25, 2025 · Jun 26, 2025 · Jul 31, 2025
diff --git a/CITATION.cff b/CITATION.cff
@@ -25,7 +25,8 @@ authors:
     orcid: 'https://orcid.org/0000-0002-3033-674X'
   - given-names: 'Shalin B.'
     family-names: Mehta
-    affiliation: [email protected]
+    email: [email protected]
+    affiliation: Chan Zuckerberg Biohub San Francisco
     orcid: 'https://orcid.org/0000-0002-2542-3582'
 repository-code: 'https://github.com/czbiohub-sf/shrimPy'
 abstract: >-

diff --git a/mantis/acquisition/AcquisitionSettings.py b/mantis/acquisition/AcquisitionSettings.py
@@ -73,6 +73,7 @@ class ChannelSettings:
     # dictionaries with following structure: {well_id: list_of_exposure_times}
     exposure_times_per_well: Dict = field(init=None, default_factory=dict)
     laser_powers_per_well: Dict = field(init=None, default_factory=dict)
+    min_exposure_time: NonNegativeFloat = 0  # in ms
 
     def __post_init__(self):
         self.num_channels = len(self.channels)

diff --git a/mantis/acquisition/acq_engine.py b/mantis/acquisition/acq_engine.py
@@ -24,6 +24,7 @@
 
 from mantis import get_console_formatter
 from mantis.acquisition import microscope_operations
+from mantis.acquisition.autoexposure import load_manual_illumination_settings
 from mantis.acquisition.hook_functions import globals
 from mantis.acquisition.logger import configure_debug_logger, log_conda_environment
 
@@ -626,6 +627,7 @@ def update_ls_acquisition_rates(self, ls_exposure_times: list):
             self.ls_acq.slice_settings.acquisition_rate = [
                 np.minimum(30, 1000 / exp_time) for exp_time in ls_exposure_times
             ]
+            self.ls_acq.channel_settings.min_exposure_time = 2  # useful for debugging
             return
 
         # Determine light-sheet acq timing
@@ -635,6 +637,9 @@ def update_ls_acquisition_rates(self, ls_exposure_times: list):
             decimals=3,
         )
         _cam_max_fps = int(np.around(1000 / ls_readout_time_ms))
+        # When using simulated global shutter by modulating the laser excitation time,
+        # the exposure time needs to be greater than the sensor readout time
+        self.ls_acq.channel_settings.min_exposure_time = ls_readout_time_ms
         for ls_exp_time in ls_exposure_times:
             assert (
                 ls_readout_time_ms < ls_exp_time
@@ -777,21 +782,49 @@ def setup_autoexposure(self):
                 self.ls_acq.channel_settings.default_laser_powers
             )
 
+        if not any(self.ls_acq.channel_settings.use_autoexposure):
+            logger.debug(
+                'Autoexposure is not enabled for any channels. Using default exposure time and laser power'
+            )
+            return
+
         if self._demo_run:
             logger.debug(
                 'Autoexposure is not supported in demo mode. Using default exposure time and laser power'
             )
             return
 
-        if (
-            any(self.ls_acq.channel_settings.use_autoexposure)
-            and self.ls_acq.autoexposure_settings.autoexposure_method == 'manual'
-        ):
+        if self.ls_acq.autoexposure_settings.autoexposure_method is None:
+            raise ValueError(
+                'Autoexposure is requested, but autoexposure settings are not provided. '
+                'Please provide autoexposure settings in the acquisition config file.'
+            )
+
+        logger.debug('Setting up autoexposure for light-sheet acquisition')
+        if self.ls_acq.autoexposure_settings.autoexposure_method == 'manual':
             # Check that the 'illumination.csv' file exists
             if not (self._root_dir / 'illumination.csv').exists():
                 raise FileNotFoundError(
                     f'The illumination.csv file required for manual autoexposure was not found in {self._root_dir}'
                 )
+            illumination_settings = load_manual_illumination_settings(
+                self._root_dir / 'illumination.csv',
+            )
+            # Check that exposure times are greater than the minimum exposure time
+            if not (
+                illumination_settings["exposure_time_ms"]
+                > self.ls_acq.channel_settings.min_exposure_time
+            ).all():
+                raise ValueError(
+                    f'All exposure times in the illumination.csv file must be greater than the minimum exposure time of {self.ls_acq.channel_settings.min_exposure_time} ms.'
+                )
+            # Check that illumination settings are provided for all wells
+            if not set(illumination_settings.index.values) == set(
+                self.position_settings.well_ids
+            ):
+                raise ValueError(
+                    'Well IDs in the illumination.csv file do not match the well IDs in the position settings.'
+                )
 
         # initialize lasers
         for channel_idx, config_name in enumerate(self.ls_acq.channel_settings.channels):

diff --git a/mantis/acquisition/autoexposure.py b/mantis/acquisition/autoexposure.py
@@ -12,34 +12,54 @@
 # adjusting camera exposure and laser power based on these suggestions until
 # convergence.
 
-import csv
-
 import numpy as np
+import pandas as pd
 
 from mantis import logger
 from mantis.acquisition.AcquisitionSettings import AutoexposureSettings
 
 
+def load_manual_illumination_settings(csv_filepath: str) -> pd.DataFrame:
+    """
+    Import the manual illumination settings from a CSV file.
+    The CSV file should have the following columns:
+    - well_id
+    - exposure_time_ms
+    - laser_power_mW
+    """
+
+    df = pd.read_csv(csv_filepath, dtype=str)
+    if not set(df.columns) == {"well_id", "exposure_time_ms", "laser_power_mW"}:
+        raise ValueError(
+            "CSV file must contain columns: well_id, exposure_time_ms, laser_power_mW"
+        )
+    df.set_index("well_id", inplace=True)
+    df["exposure_time_ms"] = df["exposure_time_ms"].astype(float)
+    df["laser_power_mW"] = df["laser_power_mW"].astype(float)
+
+    return df
+
+
 def manual_autoexposure(
     current_exposure_time,
     current_laser_power,
     illumination_settings_filepath,
     well_id,
 ):
-    autoexposure_flag = (
-        None  # 1: over-exposed , 0: optimally exposed, -1: under-exposed, None: error
-    )
-    suggested_exposure_time = current_exposure_time
-    suggested_laser_power = current_laser_power
-    with open(illumination_settings_filepath) as csvfile:
-        reader = csv.reader(csvfile)
-
-        for line in reader:
-            if line[0] == well_id:
-                autoexposure_flag = 0
-                suggested_exposure_time = float(line[1])
-                suggested_laser_power = float(line[2])
-                break
+    try:
+        autoexposure_flag = 0
+        illumination_settings = load_manual_illumination_settings(
+            illumination_settings_filepath
+        )
+        suggested_exposure_time = illumination_settings.loc[well_id, "exposure_time_ms"]
+        suggested_laser_power = illumination_settings.loc[well_id, "laser_power_mW"]
+    except Exception as e:
+        logger.error(f"Error reading manual illumination settings: {e}")
+        # If autoexposure fails, we return None for autoexposure_flag
+        # and keep the current exposure time and laser power
+        autoexposure_flag = None
+        suggested_exposure_time = current_exposure_time
+        suggested_laser_power = current_laser_power
 
     return autoexposure_flag, suggested_exposure_time, suggested_laser_power
 

diff --git a/mantis/cli/main.py b/mantis/cli/main.py
@@ -1,6 +1,7 @@
 import click
 
 from mantis.cli.run_acquisition import run_acquisition
+from mantis.cli.stir_plate import stir_plate_cli
 
 CONTEXT = {"help_option_names": ["-h", "--help"]}
 
@@ -17,3 +18,4 @@ def cli():
 
 
 cli.add_command(run_acquisition)
+cli.add_command(stir_plate_cli)
diff --git a/mantis/cli/stir_plate.py b/mantis/cli/stir_plate.py
@@ -0,0 +1,78 @@
+import logging
+import time
+
+import click
+
+from pycromanager import Core, Studio
+
+from mantis.acquisition.acq_engine import LF_ZMQ_PORT
+from mantis.acquisition.microscope_operations import get_position_list, set_xy_position
+
+logger = logging.getLogger(__name__)
+
+
+def stir_plate(duration_hours: float, dwell_time_min: int) -> None:
+    """Move through all positions defined in the Micro-manager position list,
+    dwelling at each position for dwell_time_min until duration_hours is reached.
+    This helps distribute the silicone oil evenly across the plate.
+
+    Parameters
+    ----------
+    duration_hours : float
+        Total duration for stirring the plate in hours.
+    dwell_time_min : int
+        Time spent at each position in minutes.
+    """
+
+    # Connect to the Micro-manager Studio
+    mmc = Core(port=LF_ZMQ_PORT)
+    mmStudio = Studio(port=LF_ZMQ_PORT)
+    z_stage = None
+
+    # Import positions from the Micro-manager position list
+    positions, position_labels = get_position_list(mmStudio, z_stage)
+    num_positions = len(positions)
+    if num_positions == 0:
+        raise RuntimeError(
+            "No positions found in the Micro-manager position list. "
+            "Please create a position list before running this command."
+        )
+
+    p_idx = 0
+    t_start = time.time()
+    t_end = t_start + duration_hours * 3600  # Convert hours to seconds
+    while time.time() < t_end:
+        # Move to the next position
+        logger.info(f"Moving to position: {position_labels[p_idx]}")
+        set_xy_position(mmc, positions[p_idx][:2])
+
+        # Dwell for specified amount of time
+        time.sleep(dwell_time_min * 60)
+
+        # Increment position index, wrap around if necessary
+        p_idx = (p_idx + 1) % num_positions
+
+    # Move back to the first position
+    logger.info("Stirring completed, moving to the first position.")
+    set_xy_position(mmc, positions[0][:2])
+
+
+@click.command("stir-plate")
+@click.option(
+    "--duration-hours",
+    type=float,
+    required=True,
+    help="Total duration for stirring the plate in hours.",
+)
+@click.option(
+    "--dwell-time-min",
+    type=int,
+    required=False,
+    default=1,
+    help="Time spent at each position in minutes, by default 1 minute.",
+)
+def stir_plate_cli(duration_hours: float, dwell_time_min: int) -> None:
+    """Stir the plate by moving through all positions in the
+    Micro-manager position list.
+    """
+    stir_plate(duration_hours, dwell_time_min)