Merge pull request #1012 from TheDeanLab/1008-laser-digital-only-modulation-doesnt-work-1

AdvancedImagingUTSW · web-flow · commit a69c03490013 · 2024-10-08T18:30:49.000-05:00
diff --git a/src/navigate/model/device_startup_functions.py b/src/navigate/model/device_startup_functions.py
@@ -1238,20 +1238,41 @@ def start_lasers(
     if plugin_devices is None:
         plugin_devices = {}
 
+    analog, digital, modulation = None, None, None
+
     if is_synthetic:
         device_type = "SyntheticLaser"
 
     else:
-        device_type = configuration["configuration"]["microscopes"][microscope_name][
+        analog = configuration["configuration"]["microscopes"][microscope_name][
             "lasers"
-        ][id]["power"]["hardware"]["type"]
+        ][id]["power"]["hardware"].get("type", None)
 
-    if device_type == "NI":
+        digital = configuration["configuration"]["microscopes"][microscope_name][
+            "lasers"
+        ][id]["onoff"]["hardware"].get("type", None)
+
+        device_type = analog
+
+    if analog == "NI" or digital == "NI":
         if device_connection is not None:
             return device_connection
         from navigate.model.devices.lasers.ni import LaserNI
 
-        return LaserNI(microscope_name, device_connection, configuration, id)
+        if analog == "NI" and digital == "NI":
+            modulation = "mixed"
+        elif analog == "NI":
+            modulation = "analog"
+        elif digital == "NI":
+            modulation = "digital"
+
+        return LaserNI(
+            microscope_name=microscope_name,
+            device_connection=device_connection,
+            configuration=configuration,
+            laser_id=id,
+            modulation_type=modulation,
+        )
 
     elif device_type.lower() == "syntheticlaser" or device_type.lower() == "synthetic":
         if device_connection is not None:
diff --git a/src/navigate/model/devices/lasers/ni.py b/src/navigate/model/devices/lasers/ni.py
@@ -61,6 +61,7 @@ def __init__(
         device_connection: Any,
         configuration: Dict[str, Any],
         laser_id: int,
+        modulation_type="digital",
     ) -> None:
         """Initialize the LaserNI class.
 
@@ -74,13 +75,74 @@ def __init__(
             The device configuration.
         laser_id : int
             The laser id.
+        modulation_type : str
+            The modulation type of the laser - Analog, Digital, or Mixed.
         """
         super().__init__(microscope_name, device_connection, configuration, laser_id)
 
+        #: str: The modulation type of the laser - Analog, Digital, or Mixed.
+        self.modulation_type = modulation_type
+
         #: str: Modulation type of the laser - Analog or Digital.
-        self.on_off_type = None
+        self.digital_port_type = None
+
+        #: float: The minimum digital modulation voltage.
+        self.laser_min_do = None
+
+        #: float: The maximum digital modulation voltage.
+        self.laser_max_do = None
+
+        #: nidaqmx.Task: The laser digital modulation task.
+        self.laser_do_task = None
+
+        #: float: The minimum analog modulation voltage.
+        self.laser_min_ao = None
+
+        #: float: The maximum analog modulation voltage.
+        self.laser_max_ao = None
+
+        #: nidaqmx.Task: The laser analog modulation task.
+        self.laser_ao_task = None
+
+        #: float: Current laser intensity.
+        self._current_intensity = 0
+
+        # Initialize the laser modulation type.
+        if self.modulation_type == "mixed":
+            self.initialize_digital_modulation()
+            self.initialize_analog_modulation()
+            logger.info(f"{str(self)} initialized with mixed modulation.")
+
+        elif self.modulation_type == "analog":
+            self.initialize_analog_modulation()
+            logger.info(f"{str(self)} initialized with analog modulation.")
+
+        elif self.modulation_type == "digital":
+            self.initialize_digital_modulation()
+            logger.info(f"{str(self)} initialized with digital modulation.")
+
+    def initialize_analog_modulation(self) -> None:
+        """Initialize the analog modulation of the laser."""
+        try:
+            laser_ao_port = self.device_config["power"]["hardware"]["channel"]
+
+            #: float: The minimum analog modulation voltage.
+            self.laser_min_ao = self.device_config["power"]["hardware"]["min"]
+
+            #: float: The maximum analog modulation voltage.
+            self.laser_max_ao = self.device_config["power"]["hardware"]["max"]
+
+            #: object: The laser analog modulation task.
+            self.laser_ao_task = nidaqmx.Task()
+            self.laser_ao_task.ao_channels.add_ao_voltage_chan(
+                laser_ao_port, min_val=self.laser_min_ao, max_val=self.laser_max_ao
+            )
+        except DaqError as e:
+            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}")
 
-        # Digital out (if using mixed modulation mode)
+    def initialize_digital_modulation(self) -> None:
+        """Initialize the digital modulation of the laser."""
         try:
             laser_do_port = self.device_config["onoff"]["hardware"]["channel"]
 
@@ -94,18 +156,18 @@ def __init__(
             self.laser_do_task = nidaqmx.Task()
 
             if "/ao" in laser_do_port:
-                # Artificial Digital Modulation via an Analog Port
+                # Perform the digital modulation with an analog output port.
                 self.laser_do_task.ao_channels.add_ao_voltage_chan(
                     laser_do_port, min_val=self.laser_min_do, max_val=self.laser_max_do
                 )
-                self.on_off_type = "analog"
+                self.digital_port_type = "analog"
 
             else:
                 # Digital Modulation via a Digital Port
                 self.laser_do_task.do_channels.add_do_chan(
                     laser_do_port, line_grouping=LineGrouping.CHAN_FOR_ALL_LINES
                 )
-                self.on_off_type = "digital"
+                self.digital_port_type = "digital"
         except (KeyError, DaqError) as e:
             self.laser_do_task = None
             if isinstance(e, DaqError):
@@ -115,36 +177,16 @@ def __init__(
                 print(e.error_code)
                 print(e.error_type)
 
-        #: float: Current laser intensity.
-        self._current_intensity = 0
-
-        # Analog out
-        try:
-            laser_ao_port = self.device_config["power"]["hardware"]["channel"]
-
-            #: float: The minimum analog modulation voltage.
-            self.laser_min_ao = self.device_config["power"]["hardware"]["min"]
-
-            #: float: The maximum analog modulation voltage.
-            self.laser_max_ao = self.device_config["power"]["hardware"]["max"]
-
-            #: object: The laser analog modulation task.
-            self.laser_ao_task = nidaqmx.Task()
-            self.laser_ao_task.ao_channels.add_ao_voltage_chan(
-                laser_ao_port, min_val=self.laser_min_ao, max_val=self.laser_max_ao
-            )
-        except DaqError as e:
-            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: float) -> None:
-        """Sets the laser power.
+        """Sets the analog laser power.
 
         Parameters
         ----------
         laser_intensity : float
             The laser intensity.
         """
+        if self.laser_ao_task is None:
+            return
         try:
             scaled_laser_voltage = (int(laser_intensity) / 100) * self.laser_max_ao
             self.laser_ao_task.write(scaled_laser_voltage, auto_start=True)
@@ -154,34 +196,40 @@ def set_power(self, laser_intensity: float) -> None:
 
     def turn_on(self) -> None:
         """Turns on the laser."""
+        self.set_power(self._current_intensity)
+
+        if self.laser_do_task is None:
+            return
         try:
-            self.set_power(self._current_intensity)
-            if self.laser_do_task is not None:
-                if self.on_off_type == "digital":
-                    self.laser_do_task.write(True, auto_start=True)
-                elif self.on_off_type == "analog":
-                    self.laser_do_task.write(self.laser_max_do, auto_start=True)
+            if self.digital_port_type == "digital":
+                self.laser_do_task.write(True, auto_start=True)
+            elif self.digital_port_type == "analog":
+                self.laser_do_task.write(self.laser_max_do, auto_start=True)
         except DaqError as e:
             logger.exception(e)
 
     def turn_off(self) -> None:
         """Turns off the laser."""
+        # set ao power to zero
+        tmp = self._current_intensity
+        self.set_power(0)
+        self._current_intensity = tmp
+
+        if self.laser_do_task is None:
+            return
         try:
-            tmp = self._current_intensity
-            self.set_power(0)
-            self._current_intensity = tmp
-            if self.laser_do_task is not None:
-                if self.on_off_type == "digital":
-                    self.laser_do_task.write(False, auto_start=True)
-                elif self.on_off_type == "analog":
-                    self.laser_do_task.write(self.laser_min_do, auto_start=True)
+            if self.digital_port_type == "digital":
+                self.laser_do_task.write(False, auto_start=True)
+            elif self.digital_port_type == "analog":
+                self.laser_do_task.write(self.laser_min_do, auto_start=True)
         except DaqError as e:
             logger.exception(e)
 
     def close(self) -> None:
         """Close the NI Task before exit."""
         try:
-            self.laser_ao_task.close()
+            if self.laser_ao_task is not None:
+                self.laser_ao_task.close()
             if self.laser_do_task is not None:
                 self.laser_do_task.close()
         except DaqError as e:
diff --git a/src/navigate/model/microscope.py b/src/navigate/model/microscope.py
@@ -746,6 +746,10 @@ def prepare_next_channel(self, update_daq_task_flag=True):
         self.lasers[str(self.laser_wavelength[self.current_laser_index])].set_power(
             channel["laser_power"]
         )
+        logger.info(
+            f"{self.laser_wavelength[self.current_laser_index]} "
+            f"nm laser power set to {channel['laser_power']}"
+        )
         # self.lasers[str(self.laser_wavelength[self.current_laser_index])].turn_on()
 
         # stop daq before writing new waveform
diff --git a/test/model/devices/lasers/test_laser_ni.py b/test/model/devices/lasers/test_laser_ni.py
@@ -50,6 +50,7 @@ def setUp(self) -> None:
                 device_connection=self.device_connection,
                 configuration=self.configuration,
                 laser_id=laser_id,
+                modulation_type="mixed",
             )
 
     def tearDown(self):
@@ -67,11 +68,11 @@ def test_set_power(self):
         assert self.laser._current_intensity == self.current_intensity
 
     def test_turn_on(self):
-        self.laser.on_off_type = "digital"
+        self.laser.digital_port_type = "digital"
         self.laser.turn_on()
         self.laser.laser_do_task.write.assert_called_with(True, auto_start=True)
 
-        self.laser.on_off_type = "analog"
+        self.laser.digital_port_type = "analog"
         self.laser.turn_on()
         self.laser.laser_do_task.write.assert_called_with(
             self.laser.laser_max_do, auto_start=True
@@ -81,13 +82,13 @@ def test_turn_off(self):
         self.current_intensity = random.randint(1, 100)
         self.laser._current_intensity = self.current_intensity
 
-        self.laser.on_off_type = "digital"
+        self.laser.digital_port_type = "digital"
         self.laser.turn_off()
         self.laser.laser_do_task.write.assert_called_with(False, auto_start=True)
 
         assert self.laser._current_intensity == self.current_intensity
 
-        self.laser.on_off_type = "analog"
+        self.laser.digital_port_type = "analog"
         self.laser.turn_off()
         self.laser.laser_do_task.write.assert_called_with(
             self.laser.laser_min_do, auto_start=True
