Enable Synthetic DAQ Support for Real Camera Devices

src/navigate/config/preload_rules/experiment.py

@@ -271,6 +271,10 @@ def repair_camera_parameters(context: PreloadContext) -> None:
             if camera_setting_dict[k] < 1:
                 camera_setting_dict[k] = camera_parameters_dict_sample[k]
 
+        # remove camera trigger backup before loading
+        if "trigger_source_backup" in camera_setting_dict:
+            del camera_setting_dict["trigger_source_backup"]
+
     # # cleanup
     # for k in configuration["experiment"]["CameraParameters"].keys():
     #     if k in camera_parameters_dict_sample.keys():

src/navigate/controller/controller.py

@@ -413,6 +413,10 @@ def use_asi_model(self) -> bool:
         ValueError
             If the DAQ type is unknown.
         """
+        # run in synthetic mode
+        if self.args.synthetic_hardware:
+            return False
+
         microscope_name = self.configuration["experiment"]["MicroscopeState"][
             "microscope_name"
         ]
@@ -1465,6 +1469,8 @@ def execute(self, command: str, *args: Any) -> Any:
             """
             self.sloppy_stop()
             self.update_experiment_setting()
+            # restore camera triggers in the experiment file
+            self._restore_camera_trigger_setting()
             file_directory = os.path.join(get_navigate_path(), "config")
             for config_name, filename in [
                 ("experiment", "experiment.yml"),
@@ -2057,3 +2063,35 @@ def register_event_listeners(
         """
         for event_name, event_handler in events.items():
             self.register_event_listener(event_name, event_handler)
+
+    def _restore_camera_trigger_setting(self):
+        """Restore the camera trigger setting."""
+        for microscope_name in self.configuration.get("experiment", {}).get(
+            "CameraParameters", {}
+        ):
+            if (
+                "trigger_source_backup"
+                in self.configuration["experiment"]["CameraParameters"][microscope_name]
+            ):
+                if (
+                    self.configuration["experiment"]["CameraParameters"][
+                        microscope_name
+                    ].get("trigger_source_backup")
+                    is not None
+                ):
+                    self.configuration["experiment"]["CameraParameters"][
+                        microscope_name
+                    ]["trigger_source"] = self.configuration["experiment"][
+                        "CameraParameters"
+                    ][
+                        microscope_name
+                    ][
+                        "trigger_source_backup"
+                    ]
+                else:
+                    del self.configuration["experiment"]["CameraParameters"][
+                        microscope_name
+                    ]["trigger_source"]
+                del self.configuration["experiment"]["CameraParameters"][
+                    microscope_name
+                ]["trigger_source_backup"]

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

@@ -489,3 +489,6 @@ def get_temperature(self) -> Optional[float]:
         if self.camera_parameters["cooling"] is False:
             logger.warning(f"{str(self)} does not support cooling.")
             return None
+
+    def generate_new_frame(self) -> None:
+        pass

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

@@ -132,9 +132,11 @@ def __init__(
         # Values are pulled from the CameraParameters section of the configuration.yml
         # file. Exposure time converted here from milliseconds to seconds.
 
+        self.trigger_source = "Internal"
+
         self.set_trigger_mode()
 
-        self.camera_parameters["supported_trigger_sources"] = ["External", "Internal"]
+        self.camera_parameters["supported_trigger_sources"] = ["External", "Internal", "Software"]
         self.camera_parameters["cooling"] = True
 
         cooling_state = self.camera_controller.get_property_value("cooling")
@@ -229,14 +231,16 @@ def set_trigger_mode(self, trigger_source: str = "External") -> None:
         trigger_source : str
             Trigger source: "External" or "Internal". Default is "External".
         """
-        if trigger_source == "Internal":
+        if trigger_source == "Software":
+            self.trigger_source = "Software"
             # Standard trigger mode
             self.camera_controller.set_property_value("trigger_mode", 1)
             # Internal trigger.
-            self.camera_controller.set_property_value("trigger_source", 1)
-            logger.debug("Set camera trigger mode: Free running mode.")
+            self.camera_controller.set_property_value("trigger_source", 3)
+            logger.debug("Set camera trigger mode: Software Trigger.")
 
-        else:
+        elif trigger_source == "External":
+            self.trigger_source = "External"
             self.camera_controller.set_property_value(
                 "defect_correct_mode", self.camera_parameters["defect_correct_mode"]
             )
@@ -247,6 +251,13 @@ def set_trigger_mode(self, trigger_source: str = "External") -> None:
                 "trigger_source", 2  # External trigger.
             )
             logger.debug("Set camera trigger mode: External Edge Trigger.")
+        else:
+            self.trigger_source = "Internal"
+            # Standard trigger mode
+            self.camera_controller.set_property_value("trigger_mode", 1)
+            # Internal trigger.
+            self.camera_controller.set_property_value("trigger_source", 1)
+            logger.debug("Set camera trigger mode: Free running mode.")
 
     def set_sensor_mode(self, mode: str) -> None:
         """Set HamamatsuOrca sensor mode.
@@ -515,6 +526,12 @@ def get_new_frame(self) -> list[int]:
             Frame ids from HamamatsuOrca camera.
         """
         return self.camera_controller.get_frames()
+
+    def generate_new_frame(self) -> None:
+        if not self.is_acquiring:
+            return
+        if self.trigger_source == "Software":
+            self.camera_controller.fire_software_trigger()
 
     def set_cooling(self, cooling: str) -> None:
         """Set camera cooling mode and temperature.

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

@@ -169,10 +169,17 @@ def __init__(
         self.camera_parameters["x_pixels"] = self.camera_controller.sensor_size[0]
         self.camera_parameters["y_pixels"] = self.camera_controller.sensor_size[1]
 
+        #: trigger source
+        self.trigger_source = None
+        self.camera_parameters["supported_trigger_sources"] = [
+            "External",
+            "Internal",
+            "Software",
+        ]
+        self.set_trigger_mode("External")
+
         self.set_sensor_mode("Normal")
         self.camera_controller.binning = 1
-        # TODO: Implement readout_speed, defect_correct_mode
-        self.camera_controller.exp_mode = "Edge Trigger"
         self.camera_controller.prog_scan_dir = 0
 
         # Photometrics camera settings from config file
@@ -291,12 +298,17 @@ def set_trigger_mode(self, trigger_source: str = "External") -> None:
         trigger_source : str
             Trigger source. Options are 'External' or 'Internal'.
         """
+        self.trigger_source = trigger_source
         if trigger_source == "External":
-            self.camera_controller.trig_mode = "External"
-        elif trigger_source == "Internal":
-            self.camera_controller.trig_mode = "Internal"
+            self.camera_controller.exp_mode = "Edge Trigger"
+            logger.debug("Set camera trigger mode: External Edge Trigger.")
+        elif trigger_source == "Software":
+            self.camera_controller.exp_mode = "Software Trigger Edge"
+            logger.debug("Set camera trigger mode: Software Trigger.")
         else:
-            logger.debug("Camera trigger mode not supported")
+            self.camera_controller.exp_mode = "Internal Trigger"
+            self.trigger_source = "Internal"
+            logger.debug("Set camera trigger mode: Free running mode.")
 
     def set_readout_direction(self, mode: str) -> None:
         """Set Photometrics readout direction.
@@ -542,14 +554,12 @@ def initialize_image_series(
         self._scan_mode = self.camera_controller.prog_scan_mode
         if self._scan_mode == 1:
             # Programmable scan mode (ASLM)
-            self.camera_controller.exp_mode = "Edge Trigger"
             self.camera_controller.prog_scan_line_delay = self._scan_delay
             self.camera_controller.exp_out_mode = 4
 
         else:
             # Normal mode
             self.camera_controller.exp_out_mode = "Any Row"
-            self.camera_controller.exp_mode = "Edge Trigger"
 
         # Prepare for buffered acquisition
         #: int: Number of frames
@@ -619,6 +629,12 @@ def get_new_frame(self) -> list[int]:
         """
         return self._receive_images()
 
+    def generate_new_frame(self) -> None:
+        if not self.is_acquiring:
+            return
+        if self.trigger_source == "Software":
+            self.camera_controller.sw_trigger()
+
     def close_image_series(self) -> None:
         """Close Photometrics image series.
 

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

@@ -94,18 +94,11 @@ def __init__(
         self.offset_y_step = self.cam.get_param("offsetY:inc")
         self.minimum_exposure_time = self.cam.get_param("exposure:min")
 
-        # set external rising edge trigger: XI_TRG_EDGE_RISING
-        input_trigger_port = int(self.camera_parameters.get("input_trigger_port", 2))
-        if input_trigger_port >= 1 and input_trigger_port <= 12:
-            ximea_trigger_port = f"XI_GPI_PORT{input_trigger_port}"
-        else:
-            ximea_trigger_port = "XI_GPI_PORT1"
-        self.cam.set_param("gpi_selector", ximea_trigger_port)
-        self.cam.set_param("gpi_mode", "XI_GPI_TRIGGER")
-        self.cam.set_param("trigger_source", "XI_TRG_EDGE_RISING")
+        self.set_trigger_mode("External")
 
         self.camera_parameters["supported_sensor_modes"] = ["Normal"]
         self.camera_parameters["supported_readout_directions"] = ["Top-to-Bottom"]
+        self.camera_parameters["supported_trigger_sources"] = ["External", "Internal", "Software"]
 
     def __str__(self) -> str:
         """Return string representation of Ximea Base class
@@ -214,9 +207,28 @@ def set_trigger_mode(self, trigger_source="External") -> None:
         Parameters
         ----------
         trigger_source : str
-            'External' or 'Internal'
+            'External', 'Internal' or "Software
         """
-        pass
+        self.trigger_source = trigger_source
+        if trigger_source == "Software":
+            self.cam.set_param('trigger_source', "XI_TRG_SOFTWARE")
+            logger.debug("Set camera trigger mode: Software Trigger.")
+        elif trigger_source == "External":
+            # set external rising edge trigger: XI_TRG_EDGE_RISING
+            input_trigger_port = int(self.camera_parameters.get("input_trigger_port", 1))
+            if input_trigger_port >= 1 and input_trigger_port <= 12:
+                input_trigger_port = f"XI_GPI_PORT{input_trigger_port}"
+            else:
+                input_trigger_port = "XI_GPI_PORT1"
+            self.cam.set_param("gpi_selector", input_trigger_port)
+            self.cam.set_param("gpi_mode", "XI_GPI_TRIGGER")
+            self.cam.set_param("trigger_source", "XI_TRG_EDGE_RISING")
+            logger.debug("Set camera trigger mode: External Edge Trigger.")
+        else: #Internal
+            self.trigger_source == "Internal"
+            self.cam.set_param('trigger_source', "XI_TRG_OFF")
+            logger.debug("Set camera trigger mode: Free running mode.")
+
 
     def set_readout_direction(self, mode: str) -> None:
         """Set readout direction
@@ -466,6 +478,15 @@ def get_new_frame(self) -> list[int]:
 
         return frames_received
 
+    def generate_new_frame(self) -> None:
+        """Generate an internal software trigger"""
+        if not self.is_acquiring:
+            return
+
+        if self.trigger_source == "Software":
+            # TODO: verify if it returns an error code or XI_OK
+            self.cam.set_param("trigger_software", 1)
+
 
 class MU196XRCamera(XimeaBase):
     """Ximea MU196XR class.
@@ -519,3 +540,23 @@ def __str__(str) -> str:
             String representation of Ximea MU196XR class.
         """
         return "Ximea MU196XR Camera"
+
+    def set_trigger_mode(self, trigger_source="External") -> None:
+        """Set Ximea trigger mode.
+
+        Parameters
+        ----------
+        trigger_source : str
+            'External', 'Internal' or "Software
+        """
+        if trigger_source == "External":
+            self.trigger_source = "External"
+            # on MU196 series, the trigger input is on GPI port 3, gpi index is 2
+            self.cam.set_param("gpi_selector", "XI_GPI_PORT2")
+            # need to reset the trigger mode to XI_GPI_TRIGGER, otherwise the trigger mode is XI_GPI_OFF
+            self.cam.set_param("gpi_mode", "XI_GPI_TRIGGER")
+            # set external rising edge trigger: XI_TRG_EDGE_RISING
+            self.cam.set_param("trigger_source", "XI_TRG_EDGE_RISING")
+            logger.debug("Set camera trigger mode: External Edge Trigger.")
+        else:
+            super().set_trigger_mode(trigger_source)

src/navigate/model/microscope.py

@@ -369,6 +369,17 @@ def __init__(
         # connect daq and camera in synthetic mode
         if self.daq is not None and type(self.daq).__name__ == "SyntheticDAQ":
             self.daq.add_camera(self.microscope_name, self.camera)
+            # set default trigger mode to software if daq is synthetic
+            self.configuration["experiment"]["CameraParameters"][self.microscope_name][
+                "trigger_source_backup"
+            ] = self.configuration["experiment"]["CameraParameters"][
+                self.microscope_name
+            ].get(
+                "trigger_source"
+            )
+            self.configuration["experiment"]["CameraParameters"][self.microscope_name][
+                "trigger_source"
+            ] = "Software"
 
     def update_data_buffer(
         self, data_buffer: List[np.ndarray], number_of_frames: int

test/config/test_preload.py

@@ -73,6 +73,22 @@ def test_preload_keeps_loaded_sections_as_shared_dicts(loaded_configuration):
         assert isinstance(configuration[name], DictProxy)
 
 
+def test_preload_removes_camera_trigger_source_backups(loaded_configuration):
+    manager, configuration = loaded_configuration
+
+    preload_configuration(manager, configuration)
+
+    camera_parameters = configuration["experiment"]["CameraParameters"]
+    microscope_name = configuration["experiment"]["MicroscopeState"]["microscope_name"]
+    camera_parameters["trigger_source_backup"] = "External"
+    camera_parameters[microscope_name]["trigger_source_backup"] = "External"
+
+    preload_configuration(manager, configuration)
+
+    assert "trigger_source_backup" not in camera_parameters
+    assert "trigger_source_backup" not in camera_parameters[microscope_name]
+
+
 def test_preload_renames_lasers_without_user_facing_report(loaded_configuration):
     manager, configuration = loaded_configuration
     microscope = configuration["configuration"]["microscopes"]["Mesoscale"]

test/controller/test_controller.py

@@ -149,7 +149,7 @@ def controller(tk_root):
 
     try:
         controller.execute("exit")
-    except SystemExit:
+    except (AttributeError, SystemExit):
         pass
 
     # Tear down the controller properly
@@ -933,6 +933,17 @@ def test_execute_exit_saves_gui_configuration_to_loaded_path(tmp_path):
             "waveform_templates",
         ]
     }
+    controller.configuration["experiment"]["CameraParameters"] = {
+        "Mesoscale": {
+            "trigger_source": "Software",
+            "trigger_source_backup": "External",
+        },
+        "Synthetic Scope": {
+            "trigger_source": "Software",
+            "trigger_source_backup": "Internal",
+        },
+        "Scope Without Backup": {"trigger_source": "Software"},
+    }
     controller.sloppy_stop = MagicMock()
     controller.update_experiment_setting = MagicMock()
     controller.model = MagicMock()
@@ -954,6 +965,10 @@ def test_execute_exit_saves_gui_configuration_to_loaded_path(tmp_path):
         content_dict=controller.configuration["gui"],
         filename=controller.gui_configuration_path.name,
     )
+    camera_parameters = controller.configuration["experiment"]["CameraParameters"]
+    assert camera_parameters["Mesoscale"]["trigger_source"] == "External"
+    assert camera_parameters["Synthetic Scope"]["trigger_source"] == "Internal"
+    assert camera_parameters["Scope Without Backup"]["trigger_source"] == "Software"
 
 
 def test_execute_adaptive_optics(controller):