Add KINESIS stage + new galvo waveforms and tests

docs/source/02_user_guide/01_supported_hardware/galvo.rst

@@ -12,6 +12,9 @@ National Instruments
 
 Multiple types of galvanometers have been used, including Cambridge Technologies/Novanta, Thorlabs, and ScannerMAX. Each of these devices are externally controlled via analog signals delivered from an NI-based data acquisition card.
 
+For NI and synthetic galvos, the software supports the following waveform options:
+``sawtooth``, ``sine``, ``halfsaw``, ``quadratic``, and ``centered_cubic``.
+
 .. collapse:: Configuration File
 
     .. code-block:: yaml

docs/source/02_user_guide/01_supported_hardware/stage.rst

@@ -148,6 +148,10 @@ MFC2000
 MS2000
 ~~~~~~~
 
+.. note::
+
+    On Windows, **navigate** configures serial buffer sizes for MS2000
+    communication. On Linux/macOS this tuning step is skipped for compatibility.
 
 .. collapse:: Configuration File
 
@@ -419,7 +423,7 @@ positioning.
             stage:
               hardware:
                 -
-                  type: Thorlabs
+                  type: KIM001
                   serial_number: 74000375
                   axes: [f]
                   axes_mapping: [1]
@@ -508,6 +512,58 @@ KST101
 
 |
 
+-----------------
+
+KINESIS (Serial)
+~~~~~~~~~~~~~~~~
+
+This mode supports Thorlabs Kinesis stepper communication through
+``pylablib`` using a serial device path (for example ``/dev/ttyUSB1`` on Linux).
+
+.. note::
+
+    ``steps_per_um`` is the preferred scale parameter for KINESIS stages.
+    If omitted, **navigate** falls back to ``device_units_per_mm``.
+
+.. collapse:: Configuration File
+
+    .. code-block:: yaml
+
+      microscopes:
+        microscope_name:
+            stage:
+              hardware:
+                -
+                  type: KINESIS
+                  serial_number: /dev/ttyUSB1
+                  axes: [f]
+                  axes_mapping: [1]
+                  steps_per_um: 2008.623
+                  min: 0
+                  max: 25
+              joystick_axes: [f]
+              x_min: -10000.0
+              x_max: 10000.0
+              y_min: -10000.0
+              y_max: 10000.0
+              z_min: -10000.0
+              z_max: 10000.0
+              theta_min: 0.0
+              theta_max: 360.0
+              f_min: -10000.0
+              f_max: 10000.0
+              x_offset: 0.0
+              y_offset: 0.0
+              z_offset: 0.0
+              theta_offset: 0.0
+              f_offset: 0.0
+              flip_x: False
+              flip_y: False
+              flip_z: False
+              flip_f: False
+
+|
+
 --------------
 
 .. _galvo_stage:

src/navigate/config/configuration_database.py

@@ -182,6 +182,7 @@
     "Sutter Instruments": ("MP285", "sutter"),
     "ThorLabs KCube Inertial Device KIM001": ("KIM001", "thorlabs"),
     "ThorLabs KCube Inertial Device KST101": ("KST101", "thorlabs"),
+    "ThorLabs Kinesis Stepper Motor (Serial)": ("KINESIS", "thorlabs"),
     "Virtual Device": ("Synthetic", "synthetic"),
 }
 

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

@@ -31,6 +31,7 @@
 
 
 # Standard Imports
+import platform
 import time
 import logging
 
@@ -104,8 +105,12 @@ def connect_to_serial(
         self.serial.write_timeout = write_timeout
         self.serial.timeout = read_timeout
 
-        # set the size of the rx and tx buffers before calling open
-        self.serial.set_buffer_size(rx_size, tx_size)
+        # set_buffer_size is only available/reliable on some platforms
+        if platform.system() == "Windows" and hasattr(self.serial, "set_buffer_size"):
+            try:
+                self.serial.set_buffer_size(rx_size, tx_size)
+            except Exception as e:
+                logger.debug(f"Unable to set serial buffer size on Windows: {e}")
         try:
             self.serial.open()
         except SerialException:

src/navigate/model/devices/APIs/thorlabs/pykinesis_controller.py

@@ -0,0 +1,134 @@
+# Copyright (c) 2021-2025  The University of Texas Southwestern Medical Center.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted for academic and research use only (subject to the
+# limitations in the disclaimer below) provided that the following conditions are met:
+#
+#      * Redistributions of source code must retain the above copyright notice,
+#      this list of conditions and the following disclaimer.
+#
+#      * Redistributions in binary form must reproduce the above copyright
+#      notice, this list of conditions and the following disclaimer in the
+#      documentation and/or other materials provided with the distribution.
+#
+#      * Neither the name of the copyright holders nor the names of its
+#      contributors may be used to endorse or promote products derived from this
+#      software without specific prior written permission.
+#
+# NO EXPRESS OR IMPLIED LICENSES TO ANY PARTY'S PATENT RIGHTS ARE GRANTED BY
+# THIS LICENSE. THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND
+# CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
+# PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
+# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+# BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER
+# 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.
+
+"""Thorlabs Kinesis controller wrapper backed by pylablib."""
+
+import logging
+from time import sleep
+from typing import Any
+
+# Logger Setup
+p = __name__.split(".")[1]
+logger = logging.getLogger(p)
+
+SLEEP_AFTER_WAIT = 0.100
+
+
+class KinesisStage:
+    """Simple wrapper around pylablib's Kinesis motor API."""
+
+    def __init__(self, dev_path: str, verbose: bool = False):
+        self.verbose = verbose
+        self.dev_path = str(dev_path)
+        self.stage = None
+        self.move_params: dict[str, float] = {
+            "min_velocity": 0.0,
+            "max_velocity": 0.0,
+            "acceleration": 0.0,
+        }
+        self.open()
+
+    @staticmethod
+    def _load_thorlabs_backend() -> Any:
+        """Load pylablib Thorlabs backend lazily."""
+        try:
+            from pylablib.devices import Thorlabs
+        except ImportError as e:
+            raise ImportError(
+                "pylablib is required for KINESIS stage support."
+            ) from e
+        return Thorlabs
+
+    def open(self) -> None:
+        """Open the device for communications."""
+        thorlabs = self._load_thorlabs_backend()
+        connection = {"port": self.dev_path, "baudrate": 115200, "rtscts": True}
+        try:
+            self.stage = thorlabs.KinesisMotor(("serial", connection), scale="step")
+        except Exception as e:
+            raise ConnectionError(f"KINESIS stage connection failed: {e}") from e
+
+    def close(self) -> None:
+        """Disconnect and close the device."""
+        if self.stage is None:
+            return
+        try:
+            self.stage.stop()
+        except Exception:
+            pass
+        self.stage.close()
+
+    def move_to_position(
+        self, position_um: float, steps_per_um: float, wait_till_done: bool
+    ) -> None:
+        """Move to absolute position in microns."""
+        current_steps = self.stage.get_position(channel=1, scale=False)
+        target_steps = int(round(float(position_um) * float(steps_per_um)))
+        delta_steps = target_steps - int(current_steps)
+        self.stage.move_by(delta_steps, channel=1, scale=False)
+        if wait_till_done:
+            self.stage.wait_move(channel=1)
+        if SLEEP_AFTER_WAIT:
+            sleep(SLEEP_AFTER_WAIT)
+
+    def get_current_position(self, steps_per_um: float) -> float:
+        """Get current position in microns."""
+        current_steps = self.stage.get_position(channel=1, scale=False)
+        position_um = float(current_steps) / float(steps_per_um)
+        return round(position_um, 2)
+
+    def stop(self) -> None:
+        """Halt motion."""
+        self.stage.stop()
+
+    def home_stage(self) -> None:
+        """Run homing sequence."""
+        self.stage.home()
+
+    def set_velocity_params(
+        self,
+        min_velocity: float,
+        max_velocity: float,
+        acceleration: float,
+        steps_per_um: float,
+    ) -> None:
+        """Set motion profile parameters."""
+        min_velocity_steps = min_velocity * steps_per_um
+        max_velocity_steps = max_velocity * steps_per_um
+        acceleration_steps = acceleration * steps_per_um
+        self.stage.set_move_params(
+            min_velocity_steps, max_velocity_steps, acceleration_steps
+        )
+        self.move_params = {
+            "min_velocity": min_velocity_steps,
+            "max_velocity": max_velocity_steps,
+            "acceleration": acceleration_steps,
+        }

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

@@ -38,7 +38,7 @@
 # Third Party Imports
 
 # Local Imports
-from navigate.model.waveforms import sawtooth, sine_wave
+from navigate.model.waveforms import centered_cubic, quadratic, sawtooth, sine_wave
 from navigate.tools.decorators import log_initialization
 
 # # Logger Setup
@@ -220,6 +220,24 @@ def adjust(
                         duty_cycle=galvo_rising_ramp,
                         phase=self.camera_delay,
                     )
+                elif self.galvo_waveform == "centered_cubic":
+                    self.waveform_dict[channel_key] = centered_cubic(
+                        sample_rate=self.sample_rate,
+                        sweep_time=self.sweep_time,
+                        exposure=exposure_time,
+                        delay=self.camera_delay,
+                        amplitude=galvo_amplitude,
+                        offset=galvo_offset,
+                    )
+                elif self.galvo_waveform == "quadratic":
+                    self.waveform_dict[channel_key] = quadratic(
+                        sample_rate=self.sample_rate,
+                        sweep_time=self.sweep_time,
+                        exposure=exposure_time,
+                        delay=self.camera_delay,
+                        amplitude=galvo_amplitude,
+                        offset=galvo_offset,
+                    )
                 elif self.galvo_waveform == "sine":
                     self.waveform_dict[channel_key] = sine_wave(
                         sample_rate=self.sample_rate,