From 33c57236d34d0896437827c83288d58876470ac4 Mon Sep 17 00:00:00 2001 From: Michael Honaker Date: Mon, 9 Feb 2026 14:24:13 -0500 Subject: [PATCH 01/14] Updates and stuff --- software/PythonAPI/.gitignore | 134 ++++++++++++++++++ software/PythonAPI/README.md | 79 +++++++++++ .../PythonAPI/{ => examples}/usage_example.py | 4 +- .../PythonAPI/open_micro_stage/__init__.py | 5 + .../PythonAPI/open_micro_stage/__main__.py | 3 + .../api.py} | 0 .../calibration_plotter.py | 6 +- software/PythonAPI/pyproject.toml | 75 ++++++++++ software/PythonAPI/requirements.txt | 3 - 9 files changed, 300 insertions(+), 9 deletions(-) create mode 100644 software/PythonAPI/.gitignore create mode 100644 software/PythonAPI/README.md rename software/PythonAPI/{ => examples}/usage_example.py (80%) create mode 100644 software/PythonAPI/open_micro_stage/__init__.py create mode 100644 software/PythonAPI/open_micro_stage/__main__.py rename software/PythonAPI/{open_micro_stage_api.py => open_micro_stage/api.py} (100%) rename software/PythonAPI/{ => open_micro_stage}/calibration_plotter.py (94%) create mode 100644 software/PythonAPI/pyproject.toml delete mode 100644 software/PythonAPI/requirements.txt diff --git a/software/PythonAPI/.gitignore b/software/PythonAPI/.gitignore new file mode 100644 index 0000000..f4daebb --- /dev/null +++ b/software/PythonAPI/.gitignore @@ -0,0 +1,134 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +pip-wheel-metadata/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +.python-version + +# pipenv +Pipfile.lock + +# PEP 582 +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# IDE +.vscode/ +.idea/ +*.swp +*.swo +*~ + +# OS +.DS_Store +Thumbs.db diff --git a/software/PythonAPI/README.md b/software/PythonAPI/README.md new file mode 100644 index 0000000..475fc73 --- /dev/null +++ b/software/PythonAPI/README.md @@ -0,0 +1,79 @@ +# Open Micro Stage Python API + +A Python API for controlling the Open Micro Stage micro-manipulator. + +## Installation + +### Development Installation + +```bash +pip install -e . +``` + +### Production Installation + +```bash +pip install open_micro_stage +``` + +## Usage + +```python +from open_micro_stage import OpenMicroStageInterface + +# Create interface and connect +oms = OpenMicroStageInterface(show_communication=True, show_log_messages=True) +oms.connect('/dev/ttyACM0') + +# Home device +oms.home() + +# Move to position +oms.move_to(0, 0, 0, f=10) +oms.wait_for_stop() + +# Read device state +oms.read_device_state_info() +``` + +See `examples/` directory for more usage examples. + +## Requirements + +- Python >= 3.8 +- numpy +- pyserial +- colorama + +## Development + +### Install Development Dependencies + +```bash +pip install -e ".[dev]" +``` + +### Run Tests + +```bash +pytest +``` + +### Code Quality + +The project uses [Ruff](https://github.com/astral-sh/ruff) for: +- Code formatting +- Import sorting +- Linting (pycodestyle, Pyflakes, flake8-bugbear, flake8-comprehensions) + +#### Format and lint code + +```bash +ruff format . +ruff check --fix . +``` + +## License + +See LICENSE file for details. + diff --git a/software/PythonAPI/usage_example.py b/software/PythonAPI/examples/usage_example.py similarity index 80% rename from software/PythonAPI/usage_example.py rename to software/PythonAPI/examples/usage_example.py index 212a305..c458ce3 100644 --- a/software/PythonAPI/usage_example.py +++ b/software/PythonAPI/examples/usage_example.py @@ -1,4 +1,4 @@ -from open_micro_stage_api import OpenMicroStageInterface +from open_micro_stage import OpenMicroStageInterface # create interface and connect oms = OpenMicroStageInterface(show_communication=True, show_log_messages=True) @@ -15,4 +15,4 @@ oms.wait_for_stop() # print some info -oms.read_device_state_info() \ No newline at end of file +oms.read_device_state_info() diff --git a/software/PythonAPI/open_micro_stage/__init__.py b/software/PythonAPI/open_micro_stage/__init__.py new file mode 100644 index 0000000..9c52e66 --- /dev/null +++ b/software/PythonAPI/open_micro_stage/__init__.py @@ -0,0 +1,5 @@ +"""Open Micro Stage - Python API for micro-manipulator control.""" + +from .api import OpenMicroStageInterface + +__all__ = ["OpenMicroStageInterface"] diff --git a/software/PythonAPI/open_micro_stage/__main__.py b/software/PythonAPI/open_micro_stage/__main__.py new file mode 100644 index 0000000..2fc8b82 --- /dev/null +++ b/software/PythonAPI/open_micro_stage/__main__.py @@ -0,0 +1,3 @@ +"""Main file for open micro stage""" +from open_micro_stage.calibration_plotter import main +main() \ No newline at end of file diff --git a/software/PythonAPI/open_micro_stage_api.py b/software/PythonAPI/open_micro_stage/api.py similarity index 100% rename from software/PythonAPI/open_micro_stage_api.py rename to software/PythonAPI/open_micro_stage/api.py diff --git a/software/PythonAPI/calibration_plotter.py b/software/PythonAPI/open_micro_stage/calibration_plotter.py similarity index 94% rename from software/PythonAPI/calibration_plotter.py rename to software/PythonAPI/open_micro_stage/calibration_plotter.py index c0e3f26..af3d709 100644 --- a/software/PythonAPI/calibration_plotter.py +++ b/software/PythonAPI/open_micro_stage/calibration_plotter.py @@ -1,4 +1,4 @@ -from open_micro_stage_api import OpenMicroStageInterface +from open_micro_stage import OpenMicroStageInterface import matplotlib.pyplot as plt plt.rcParams['figure.dpi'] = 200 @@ -35,6 +35,4 @@ def main(): # Adjust layout and show plt.tight_layout() - plt.show() - -main() \ No newline at end of file + plt.show() \ No newline at end of file diff --git a/software/PythonAPI/pyproject.toml b/software/PythonAPI/pyproject.toml new file mode 100644 index 0000000..ebb7471 --- /dev/null +++ b/software/PythonAPI/pyproject.toml @@ -0,0 +1,75 @@ +[project] +name = "open_micro_stage" +version = "0.1.0" +description = "Python API for controlling the Open Micro Stage manipulator" +readme = "README.md" +requires-python = ">=3.8" +license = {text = "MIT"} +authors = [ + {name = "MicroManipulatorStepper Contributors"} +] +keywords = ["micromanipulator", "stepper", "stage", "control", "robotics"] +classifiers = [ + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "Intended Audience :: Science/Research", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Topic :: Scientific/Engineering", + "Topic :: Software Development :: Libraries :: Python Modules", +] +dependencies = [ + "numpy", + "pyserial", + "colorama", +] + +[build-system] +requires = ["setuptools>=65.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project.optional-dependencies] +dev = [ + "pytest>=7.0", + "pytest-cov", + "ruff", +] + +[project.urls] +Homepage = "https://github.com/HonakerM/MicroManipulatorStepper" +Repository = "https://github.com/HonakerM/MicroManipulatorStepper.git" +Documentation = "https://github.com/HonakerM/MicroManipulatorStepper" +Issues = "https://github.com/HonakerM/MicroManipulatorStepper/issues" + +[tool.setuptools] +packages = ["open_micro_stage"] +package-dir = {"" = "src"} + +[tool.setuptools.package-data] +open_micro_stage = ["py.typed"] + +[tool.ruff] +line-length = 120 +target-version = "py38" + +[tool.ruff.lint] +select = [ + "E", # pycodestyle errors + "W", # pycodestyle warnings + "F", # Pyflakes + "I", # isort + "B", # flake8-bugbear + "C4", # flake8-comprehensions + "UP", # pyupgrade +] +ignore = [ + "E501", # line too long (handled by formatter) +] + +[tool.ruff.lint.isort] +profile = "black" \ No newline at end of file diff --git a/software/PythonAPI/requirements.txt b/software/PythonAPI/requirements.txt deleted file mode 100644 index 6e4d37a..0000000 --- a/software/PythonAPI/requirements.txt +++ /dev/null @@ -1,3 +0,0 @@ -numpy -pyserial -colorama From bcaa5d21a240d0f94c88c8cfaa25e55d937d7bce Mon Sep 17 00:00:00 2001 From: Michael Honaker Date: Mon, 9 Feb 2026 14:33:55 -0500 Subject: [PATCH 02/14] Updates and stufff --- software/PythonAPI/LICENSE | 21 +++++++++++++++++++++ software/PythonAPI/pyproject.toml | 15 +-------------- 2 files changed, 22 insertions(+), 14 deletions(-) create mode 100644 software/PythonAPI/LICENSE diff --git a/software/PythonAPI/LICENSE b/software/PythonAPI/LICENSE new file mode 100644 index 0000000..14ebf02 --- /dev/null +++ b/software/PythonAPI/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2025 Github User '0x23' (https://github.com/0x23/) + +Permission is hereby granted, free of charge, to any person obtaining a copy of this +software, hardware design, documentation, or concept (the "Work"), to deal in the Work +without restriction, including without limitation the rights to use, copy, modify, merge, +publish, distribute, sublicense, and/or sell copies of the Work, and to permit persons +to whom the Work is furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all copies +or substantial portions of the Work, including but not limited to products or +derivative works based on the presented concepts, designs, or arrangements, even if +the original design files are not directly used. + +THE WORK IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, +INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE WORK +OR THE USE OR OTHER DEALINGS IN THE WORK. diff --git a/software/PythonAPI/pyproject.toml b/software/PythonAPI/pyproject.toml index ebb7471..49de14a 100644 --- a/software/PythonAPI/pyproject.toml +++ b/software/PythonAPI/pyproject.toml @@ -4,7 +4,7 @@ version = "0.1.0" description = "Python API for controlling the Open Micro Stage manipulator" readme = "README.md" requires-python = ">=3.8" -license = {text = "MIT"} +license = "MIT" authors = [ {name = "MicroManipulatorStepper Contributors"} ] @@ -13,15 +13,6 @@ classifiers = [ "Development Status :: 3 - Alpha", "Intended Audience :: Developers", "Intended Audience :: Science/Research", - "License :: OSI Approved :: MIT License", - "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.8", - "Programming Language :: Python :: 3.9", - "Programming Language :: Python :: 3.10", - "Programming Language :: Python :: 3.11", - "Programming Language :: Python :: 3.12", - "Topic :: Scientific/Engineering", - "Topic :: Software Development :: Libraries :: Python Modules", ] dependencies = [ "numpy", @@ -48,10 +39,6 @@ Issues = "https://github.com/HonakerM/MicroManipulatorStepper/issues" [tool.setuptools] packages = ["open_micro_stage"] -package-dir = {"" = "src"} - -[tool.setuptools.package-data] -open_micro_stage = ["py.typed"] [tool.ruff] line-length = 120 From a0e24d5e188170ae9a2b2cd633a9a163a2012721 Mon Sep 17 00:00:00 2001 From: Michael Honaker Date: Mon, 9 Feb 2026 14:35:30 -0500 Subject: [PATCH 03/14] Updates with ruff formatting --- software/PythonAPI/examples/usage_example.py | 2 +- .../PythonAPI/open_micro_stage/__main__.py | 4 +- software/PythonAPI/open_micro_stage/api.py | 153 ++++++++++-------- .../open_micro_stage/calibration_plotter.py | 28 ++-- software/PythonAPI/pyproject.toml | 5 +- 5 files changed, 106 insertions(+), 86 deletions(-) diff --git a/software/PythonAPI/examples/usage_example.py b/software/PythonAPI/examples/usage_example.py index c458ce3..0e8bce6 100644 --- a/software/PythonAPI/examples/usage_example.py +++ b/software/PythonAPI/examples/usage_example.py @@ -2,7 +2,7 @@ # create interface and connect oms = OpenMicroStageInterface(show_communication=True, show_log_messages=True) -oms.connect('/dev/ttyACM0') +oms.connect("/dev/ttyACM0") # run this once to calibrate joints # for i in range(3): oms.calibrate_joint(i, save_result=True) diff --git a/software/PythonAPI/open_micro_stage/__main__.py b/software/PythonAPI/open_micro_stage/__main__.py index 2fc8b82..25afde7 100644 --- a/software/PythonAPI/open_micro_stage/__main__.py +++ b/software/PythonAPI/open_micro_stage/__main__.py @@ -1,3 +1,5 @@ """Main file for open micro stage""" + from open_micro_stage.calibration_plotter import main -main() \ No newline at end of file + +main() diff --git a/software/PythonAPI/open_micro_stage/api.py b/software/PythonAPI/open_micro_stage/api.py index 3dc8835..7da6bc3 100644 --- a/software/PythonAPI/open_micro_stage/api.py +++ b/software/PythonAPI/open_micro_stage/api.py @@ -1,27 +1,27 @@ +import re import threading import time -import re from enum import Enum -import serial import numpy as np -from colorama import Fore, Style, init +import serial +from colorama import Fore, Style # --- SerialInterface -------------------------------------------------------------------------------------------------- -class SerialInterface: +class SerialInterface: class ReplyStatus(Enum): - OK = 'ok' - ERROR = 'error' - TIMEOUT = 'timeout' - BUSY = 'busy' + OK = "ok" + ERROR = "error" + TIMEOUT = "timeout" + BUSY = "busy" class LogLevel(Enum): - DEBUG = 'debug' - INFO = 'info' - WARNING = 'warning' - ERROR = 'error' + DEBUG = "debug" + INFO = "info" + WARNING = "warning" + ERROR = "error" # Static mapping from prefix to LogLevel log_level_prefix_map = { @@ -31,11 +31,15 @@ class LogLevel(Enum): "E)": LogLevel.ERROR, } - def __init__(self, port: str, baud_rate: int = 115200, - command_msg_callback=None, - log_msg_callback=None, - unsolicited_msg_callback=None, - reconnect_timeout: int = 5): + def __init__( + self, + port: str, + baud_rate: int = 115200, + command_msg_callback=None, + log_msg_callback=None, + unsolicited_msg_callback=None, + reconnect_timeout: int = 5, + ): """ Initializes the serial connection and starts background reader. :param port: Serial port name (e.g., 'COM3' or '/dev/ttyUSB0'). @@ -66,27 +70,26 @@ def __init__(self, port: str, baud_rate: int = 115200, self._reader_thread = threading.Thread(target=self._reader_loop, daemon=True) self._reader_thread.start() - def connect(self, timeout): """ Try to open the serial port. Retry until timeout expires. """ deadline = time.time() + timeout - print(Fore.MAGENTA, end='') - print(f"[SerialInterface] Connecting to port '{self.port}'...", end='') + print(Fore.MAGENTA, end="") + print(f"[SerialInterface] Connecting to port '{self.port}'...", end="") while time.time() < deadline: try: self.serial = serial.Serial(self.port, self.baud_rate, timeout=2) - print(f" [OK]") - print(Style.RESET_ALL, end='') + print(" [OK]") + print(Style.RESET_ALL, end="") return True - except (serial.SerialException, OSError) as e: - print('.', end='') + except (serial.SerialException, OSError): + print(".", end="") time.sleep(0.2) print(f" [FAILED] Timeout after {timeout} seconds.") - print(f"[SerialInterface] Connection is permanently closed") - print(Style.RESET_ALL, end='') + print("[SerialInterface] Connection is permanently closed") + print(Style.RESET_ALL, end="") self.serial = None return False @@ -98,8 +101,8 @@ def _reader_loop(self): while True: try: if self.serial is not None and self.serial.in_waiting: - char = self.serial.read(1).decode('ascii', errors='ignore') - if char in ['\n', '\r']: + char = self.serial.read(1).decode("ascii", errors="ignore") + if char in ["\n", "\r"]: if len(buffer) > 0: self._handle_line(buffer) buffer = "" @@ -108,7 +111,7 @@ def _reader_loop(self): else: time.sleep(0.001) except (serial.SerialException, OSError) as e: - print(Fore.MAGENTA+f"[SerialInterface] Lost connection: {e}"+Style.RESET_ALL) + print(Fore.MAGENTA + f"[SerialInterface] Lost connection: {e}" + Style.RESET_ALL) try: if self.serial is not None and self.serial.is_open: self.serial.close() @@ -128,7 +131,8 @@ def _handle_line(self, line: str): # print(line) # log message if log_level is not None: - if self.log_message_callback: self.log_message_callback(log_level, log_msg) + if self.log_message_callback: + self.log_message_callback(log_level, log_msg) # response elif self._waiting_for_response: line_lower = line.lower() @@ -144,15 +148,16 @@ def _handle_line(self, line: str): if self._response_status is not None: self._condition.notify() else: - self._response_string += line + '\n' + self._response_string += line + "\n" # unsolicited message else: - if self.unsolicited_msg_callback: self.unsolicited_msg_callback(line) + if self.unsolicited_msg_callback: + self.unsolicited_msg_callback(line) def _check_log_msg(self, msg: str): if len(msg) < 2: - return None, '' + return None, "" return self.log_level_prefix_map.get(msg[:2]), msg[2:] def send_command(self, cmd: str, timeout=2) -> tuple[ReplyStatus, str]: @@ -164,7 +169,7 @@ def send_command(self, cmd: str, timeout=2) -> tuple[ReplyStatus, str]: """ with self._lock: if not self.serial or not self.serial.is_open: - return SerialInterface.ReplyStatus.ERROR, 'Serial not open' + return SerialInterface.ReplyStatus.ERROR, "Serial not open" # Reset state self._waiting_for_response = True @@ -172,11 +177,11 @@ def send_command(self, cmd: str, timeout=2) -> tuple[ReplyStatus, str]: self._response_error_msg = "" self._response_status = None - cmd = (cmd.strip() + "\n") - self.command_msg_callback(cmd, None, '') + cmd = cmd.strip() + "\n" + self.command_msg_callback(cmd, None, "") # Send command - self.serial.write(cmd.encode('ascii')) + self.serial.write(cmd.encode("ascii")) self.serial.flush() # Wait for completion @@ -185,7 +190,11 @@ def send_command(self, cmd: str, timeout=2) -> tuple[ReplyStatus, str]: remaining = end_time - time.time() if remaining <= 0: self._waiting_for_response = False - print(Fore.MAGENTA + f"[SerialInterface] Command timeout, device didn't reply in time" + Style.RESET_ALL) + print( + Fore.MAGENTA + + "[SerialInterface] Command timeout, device didn't reply in time" + + Style.RESET_ALL + ) return SerialInterface.ReplyStatus.TIMEOUT, self._response_string self._condition.wait(timeout=remaining) @@ -198,12 +207,14 @@ def close(self): if self.serial and self.serial.is_open: self.serial.close() + # --- OpenMicroStageInterface ------------------------------------------------------------------------------------------ + class OpenMicroStageInterface: # Mapping log levels to colors LOG_COLORS = { - SerialInterface.LogLevel.DEBUG: Fore.WHITE+Style.DIM, + SerialInterface.LogLevel.DEBUG: Fore.WHITE + Style.DIM, SerialInterface.LogLevel.INFO: Style.RESET_ALL, SerialInterface.LogLevel.WARNING: Fore.YELLOW, SerialInterface.LogLevel.ERROR: Fore.RED, @@ -220,21 +231,27 @@ def connect(self, port: str, baud_rate: int = 921600): def version_to_str(v): return f"v{v[0]}.{v[1]}.{v[2]}" - if self.serial is not None: self.disconnect() - self.serial = SerialInterface(port, baud_rate, - log_msg_callback=self.log_msg_callback, - command_msg_callback=self.command_msg_callback, - unsolicited_msg_callback=self.unsolicited_msg_callback) + if self.serial is not None: + self.disconnect() + self.serial = SerialInterface( + port, + baud_rate, + log_msg_callback=self.log_msg_callback, + command_msg_callback=self.command_msg_callback, + unsolicited_msg_callback=self.unsolicited_msg_callback, + ) self.disable_message_callbacks = True fw_version = self.read_firmware_version() min_fw_version = (1, 0, 1) print(Fore.MAGENTA + f"Firmware version: {version_to_str(fw_version)}" + Style.RESET_ALL) if fw_version < min_fw_version: - print(Fore.MAGENTA + f"Firmware version {version_to_str(fw_version)} incompatible. " - f"At least {version_to_str(min_fw_version)} required" + Style.RESET_ALL) + print( + Fore.MAGENTA + f"Firmware version {version_to_str(fw_version)} incompatible. " + f"At least {version_to_str(min_fw_version)} required" + Style.RESET_ALL + ) self.serial = None - print('') + print("") self.disable_message_callbacks = False def disconnect(self): @@ -258,17 +275,17 @@ def command_msg_callback(self, msg, reply_status: SerialInterface.ReplyStatus, e if reply_status is not None: if msg: - msg = '\n'.join('> ' + line for line in msg.splitlines()) + msg = "\n".join("> " + line for line in msg.splitlines()) print(f"{msg.rstrip()}") if error_msg: print(f"{Style.BRIGHT}{str(reply_status.name)}:{Style.RESET_ALL} {error_msg}\n") else: print(f"{Style.BRIGHT}{str(reply_status.name)} {Style.RESET_ALL}\n") else: - print(f"{Fore.GREEN+Style.BRIGHT}{msg.rstrip()}{Style.RESET_ALL}") + print(f"{Fore.GREEN + Style.BRIGHT}{msg.rstrip()}{Style.RESET_ALL}") def unsolicited_msg_callback(self, msg): - print(Fore.CYAN+msg+Style.RESET_ALL) + print(Fore.CYAN + msg + Style.RESET_ALL) pass def set_workspace_transform(self, transform): @@ -282,8 +299,8 @@ def read_firmware_version(self): if ok != SerialInterface.ReplyStatus.OK or len(response) == 0: return 0, 0, 0 - major, minor, patch = map(int, re.match(r'v(\d+)\.(\d+)\.(\d+)', response).groups()) - return major,minor,patch + major, minor, patch = map(int, re.match(r"v(\d+)\.(\d+)\.(\d+)", response).groups()) + return major, minor, patch def home(self, axis_list=None): """ @@ -291,15 +308,15 @@ def home(self, axis_list=None): :param axis_list: Optional list of axis indices to home. If None, all axes are homed. :return: The status of the command (e.g. OK, ERROR, TIMEOUT). """ - cmd = 'G28' - axis_chars = ['A', 'B', 'C', 'D', 'E', 'F'] + cmd = "G28" + axis_chars = ["A", "B", "C", "D", "E", "F"] if axis_list is None: - axis_list = [i for i in range(len(axis_chars))] + axis_list = list(range(len(axis_chars))) for axis_idx in axis_list: if 0 > axis_idx >= len(axis_chars): - raise ValueError('Axis index out of range') - cmd += ' '+axis_chars[axis_idx] + raise ValueError("Axis index out of range") + cmd += " " + axis_chars[axis_idx] res, msg = self.serial.send_command(cmd + "\n", 10) return res @@ -315,7 +332,8 @@ def calibrate_joint(self, joint_index: int, save_result: bool): :return: """ cmd = f"M56 J{joint_index} P" - if save_result: cmd += ' S' + if save_result: + cmd += " S" res, msg = self.serial.send_command(cmd, 30) calibration_data = self._parse_table_data(msg, 3) @@ -364,11 +382,13 @@ def set_max_acceleration(self, linear_accel, angular_accel): def wait_for_stop(self, polling_interval_ms=10, disable_callbacks=True): disable_message_callbacks_prev = self.disable_message_callbacks - if disable_callbacks: self.disable_message_callbacks = True + if disable_callbacks: + self.disable_message_callbacks = True while True: res, msg = self.serial.send_command("M53\n") - if res != SerialInterface.ReplyStatus.OK: return res + if res != SerialInterface.ReplyStatus.OK: + return res elif msg.strip() == "1": return SerialInterface.ReplyStatus.OK @@ -380,10 +400,7 @@ def read_current_position(self): return None, None, None # Match values with NO space between axis letter and number - match = re.search( - r"X([-+]?\d*\.?\d+)\s*Y([-+]?\d*\.?\d+)\s*Z([-+]?\d*\.?\d+)", - response - ) + match = re.search(r"X([-+]?\d*\.?\d+)\s*Y([-+]?\d*\.?\d+)\s*Z([-+]?\d*\.?\d+)", response) if not match: raise ValueError(f"Invalid format: {response}") @@ -406,7 +423,7 @@ def set_servo_parameter(self, pos_kp=150, pos_ki=50000, vel_kp=0.2, vel_ki=100, return res def enable_motors(self, enable): - cmd = f"M17" if enable else "M18" + cmd = "M17" if enable else "M18" res, msg = self.serial.send_command(cmd, timeout=5) return res @@ -415,11 +432,11 @@ def set_pose(self, x, y, z): transformed = self.workspace_transform @ np.array([x, y, z, 1.0]) x_t, y_t, z_t = transformed[:3] / transformed[3] - cmd = f"G24 X{x_t:.6f} Y{y_t:.6f} Z{z_t:.6f}" # TODO: A, B ,C + cmd = f"G24 X{x_t:.6f} Y{y_t:.6f} Z{z_t:.6f}" # TODO: A, B ,C res, msg = self.serial.send_command(cmd) return res - def send_command(self, cmd: str, timeout_s: float=5): + def send_command(self, cmd: str, timeout_s: float = 5): res, msg = self.serial.send_command(cmd, timeout_s) return res, msg @@ -429,7 +446,7 @@ def _parse_table_data(data_string, cols): data = [[] for _ in range(cols)] for line in data_string.strip().splitlines(): - parts = line.strip().split(',') + parts = line.strip().split(",") if len(parts) != cols: continue # skip malformed lines numbers = map(float, parts) diff --git a/software/PythonAPI/open_micro_stage/calibration_plotter.py b/software/PythonAPI/open_micro_stage/calibration_plotter.py index af3d709..ff9e737 100644 --- a/software/PythonAPI/open_micro_stage/calibration_plotter.py +++ b/software/PythonAPI/open_micro_stage/calibration_plotter.py @@ -1,38 +1,42 @@ -from open_micro_stage import OpenMicroStageInterface import matplotlib.pyplot as plt -plt.rcParams['figure.dpi'] = 200 + +from open_micro_stage import OpenMicroStageInterface + +plt.rcParams["figure.dpi"] = 200 + def plot_calibration_data(ax_encoder_counts, ax_field_angel, label, data): # Plot on the provided Axes object if ax_encoder_counts is not None: ax_encoder_counts.plot(data[0], data[2], label=label) - ax_encoder_counts.set_xlabel('Motor Angle [rad]') - ax_encoder_counts.set_ylabel('Encoder Counts Raw') - ax_encoder_counts.set_title('Encoder Count Plot') + ax_encoder_counts.set_xlabel("Motor Angle [rad]") + ax_encoder_counts.set_ylabel("Encoder Counts Raw") + ax_encoder_counts.set_title("Encoder Count Plot") ax_encoder_counts.legend() ax_encoder_counts.grid(True) # Plot on the provided Axes object if ax_field_angel is not None: ax_field_angel.plot(data[0], data[1], label=label) - ax_field_angel.set_xlabel('Motor Angle [rad]') - ax_field_angel.set_ylabel('Motor Field Angle [rad]') - ax_field_angel.set_title('Field Angle Plot') + ax_field_angel.set_xlabel("Motor Angle [rad]") + ax_field_angel.set_ylabel("Motor Field Angle [rad]") + ax_field_angel.set_title("Field Angle Plot") ax_field_angel.legend() ax_field_angel.grid(True) + def main(): # create interface and connect oms = OpenMicroStageInterface(show_communication=True, show_log_messages=True) - oms.connect('/dev/ttyACM0') + oms.connect("/dev/ttyACM0") # Create subplots - fig, ax = plt.subplots(1, 1, figsize=(10, 7), sharex='all') + fig, ax = plt.subplots(1, 1, figsize=(10, 7), sharex="all") for i in range(3): res, data = oms.calibrate_joint(i, save_result=False) - plot_calibration_data(ax, None, f'Actuator {i}', data) + plot_calibration_data(ax, None, f"Actuator {i}", data) # Adjust layout and show plt.tight_layout() - plt.show() \ No newline at end of file + plt.show() diff --git a/software/PythonAPI/pyproject.toml b/software/PythonAPI/pyproject.toml index 49de14a..b2d1e19 100644 --- a/software/PythonAPI/pyproject.toml +++ b/software/PythonAPI/pyproject.toml @@ -56,7 +56,4 @@ select = [ ] ignore = [ "E501", # line too long (handled by formatter) -] - -[tool.ruff.lint.isort] -profile = "black" \ No newline at end of file +] \ No newline at end of file From 2d1c9a2f8566266fdf8d0ddd90a5df2266f17b53 Mon Sep 17 00:00:00 2001 From: Michael Honaker Date: Mon, 9 Feb 2026 15:09:58 -0500 Subject: [PATCH 04/14] test updates --- software/PythonAPI/open_micro_stage/utils.py | 34 + software/PythonAPI/pyproject.toml | 6 + software/PythonAPI/tests/__init__.py | 1 + .../PythonAPI/tests/test_open_micro_stage.py | 616 ++++++++++++++++++ 4 files changed, 657 insertions(+) create mode 100644 software/PythonAPI/open_micro_stage/utils.py create mode 100644 software/PythonAPI/tests/__init__.py create mode 100644 software/PythonAPI/tests/test_open_micro_stage.py diff --git a/software/PythonAPI/open_micro_stage/utils.py b/software/PythonAPI/open_micro_stage/utils.py new file mode 100644 index 0000000..a555a1f --- /dev/null +++ b/software/PythonAPI/open_micro_stage/utils.py @@ -0,0 +1,34 @@ + +type LocalArray = list[float] +type LocalMatrix = list[LocalArray] + +def generate_eye(size: int)->LocalMatrix: + """Generate a 2D list representing an eye pattern.""" + return [[1 if i == j else 0 for j in range(size)] for i in range(size)] + +def matrix_multiply(A: LocalMatrix, B: LocalMatrix)->LocalMatrix: + """Multiply two matrices A and B.""" + if len(A[0]) != len(B): + raise ValueError("Number of columns in A must equal number of rows in B.") + + result = [[0 for _ in range(len(B[0]))] for _ in range(len(A))] + + for i in range(len(A)): + for j in range(len(B[0])): + for k in range(len(B)): + result[i][j] += A[i][k] * B[k][j] + + return result + +def matrix_dot_product(A: LocalMatrix, B: LocalArray)->LocalArray: + """Calculate the dot product of two matrices A and B.""" + if len(A[0]) != len(B): + raise ValueError("Number of columns in A must equal number of elements in B.") + + result = [0 for _ in range(len(A))] + + for i in range(len(A)): + for j in range(len(B)): + result[i] += A[i][j] * B[j] + + return result \ No newline at end of file diff --git a/software/PythonAPI/pyproject.toml b/software/PythonAPI/pyproject.toml index b2d1e19..8eee27a 100644 --- a/software/PythonAPI/pyproject.toml +++ b/software/PythonAPI/pyproject.toml @@ -25,6 +25,12 @@ requires = ["setuptools>=65.0", "wheel"] build-backend = "setuptools.build_meta" [project.optional-dependencies] +plotter = [ + "matplotlib", +] +pretty = [ + "colorama", +] dev = [ "pytest>=7.0", "pytest-cov", diff --git a/software/PythonAPI/tests/__init__.py b/software/PythonAPI/tests/__init__.py new file mode 100644 index 0000000..cd71f58 --- /dev/null +++ b/software/PythonAPI/tests/__init__.py @@ -0,0 +1 @@ +"""Tests package for open_micro_stage.""" diff --git a/software/PythonAPI/tests/test_open_micro_stage.py b/software/PythonAPI/tests/test_open_micro_stage.py new file mode 100644 index 0000000..a3d52fd --- /dev/null +++ b/software/PythonAPI/tests/test_open_micro_stage.py @@ -0,0 +1,616 @@ +"""Tests for the OpenMicroStageInterface class.""" + +import re +import unittest +from unittest.mock import MagicMock, Mock, patch + +import numpy as np + +from open_micro_stage.api import OpenMicroStageInterface, SerialInterface + + +class TestOpenMicroStageInterface(unittest.TestCase): + """End-to-end tests for OpenMicroStageInterface.""" + + def setUp(self): + """Set up test fixtures with mocked SerialInterface.""" + # Patch SerialInterface.__init__ to prevent actual connection attempts + self.patcher = patch("open_micro_stage.api.SerialInterface.__init__", return_value=None) + self.mock_init = self.patcher.start() + + # Create a mock serial instance that we can control + self.mock_serial_instance = MagicMock(spec=SerialInterface) + + # Patch the SerialInterface class to return our mock instance + self.class_patcher = patch( + "open_micro_stage.api.SerialInterface", + return_value=self.mock_serial_instance + ) + self.mock_serial_class = self.class_patcher.start() + + # Create the interface + self.interface = OpenMicroStageInterface( + show_communication=False, show_log_messages=False + ) + + def tearDown(self): + """Clean up patches.""" + self.patcher.stop() + self.class_patcher.stop() + + def test_initialization(self): + """Test that OpenMicroStageInterface initializes with correct defaults.""" + interface = OpenMicroStageInterface() + self.assertIsNone(interface.serial) + self.assertTrue(np.array_equal(interface.workspace_transform, np.eye(4))) + self.assertTrue(interface.show_communication) + self.assertTrue(interface.show_log_messages) + self.assertFalse(interface.disable_message_callbacks) + + def test_initialization_with_params(self): + """Test initialization with custom parameters.""" + interface = OpenMicroStageInterface( + show_communication=False, show_log_messages=False + ) + self.assertFalse(interface.show_communication) + self.assertFalse(interface.show_log_messages) + + def test_connect_success(self): + """Test successful connection to device.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "v1.0.1", + ) + + self.interface.connect("/dev/ttyACM0") + + self.assertIsNotNone(self.interface.serial) + self.mock_serial_class.assert_called_once_with( + "/dev/ttyACM0", + 921600, + log_msg_callback=self.interface.log_msg_callback, + command_msg_callback=self.interface.command_msg_callback, + unsolicited_msg_callback=self.interface.unsolicited_msg_callback, + ) + + def test_connect_with_custom_baud_rate(self): + """Test connection with custom baud rate.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "v1.0.1", + ) + + self.interface.connect("/dev/ttyACM0", baud_rate=115200) + + self.mock_serial_class.assert_called_once_with( + "/dev/ttyACM0", + 115200, + log_msg_callback=self.interface.log_msg_callback, + command_msg_callback=self.interface.command_msg_callback, + unsolicited_msg_callback=self.interface.unsolicited_msg_callback, + ) + + def test_connect_incompatible_firmware(self): + """Test connection fails with incompatible firmware version.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "v0.9.0", + ) + + self.interface.connect("/dev/ttyACM0") + + # Serial should be set to None on incompatible version + self.assertIsNone(self.interface.serial) + + def test_disconnect(self): + """Test disconnection from device.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "v1.0.1", + ) + + self.interface.connect("/dev/ttyACM0") + self.interface.disconnect() + + self.mock_serial_instance.close.assert_called_once() + self.assertIsNone(self.interface.serial) + + def test_disconnect_when_not_connected(self): + """Test disconnect gracefully handles when not connected.""" + # Should not raise exception + self.interface.disconnect() + self.assertIsNone(self.interface.serial) + + def test_set_and_get_workspace_transform(self): + """Test setting and getting workspace transform.""" + transform = np.array([ + [1, 0, 0, 1], + [0, 1, 0, 2], + [0, 0, 1, 3], + [0, 0, 0, 1] + ]) + + self.interface.set_workspace_transform(transform) + result = self.interface.get_workspace_transform() + + self.assertTrue(np.array_equal(result, transform)) + + def test_read_firmware_version(self): + """Test reading firmware version.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "v1.2.3", + ) + self.interface.serial = self.mock_serial_instance + + major, minor, patch = self.interface.read_firmware_version() + + self.assertEqual(major, 1) + self.assertEqual(minor, 2) + self.assertEqual(patch, 3) + self.mock_serial_instance.send_command.assert_called_with("M58") + + def test_read_firmware_version_error(self): + """Test firmware version returns 0,0,0 on error.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.ERROR, + "", + ) + self.interface.serial = self.mock_serial_instance + + major, minor, patch = self.interface.read_firmware_version() + + self.assertEqual((major, minor, patch), (0, 0, 0)) + + def test_home_all_axes(self): + """Test homing all axes.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "", + ) + self.interface.serial = self.mock_serial_instance + + result = self.interface.home() + + self.assertEqual(result, SerialInterface.ReplyStatus.OK) + self.mock_serial_instance.send_command.assert_called_with("G28 A B C D E F\n", 10) + + def test_home_specific_axes(self): + """Test homing specific axes.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "", + ) + self.interface.serial = self.mock_serial_instance + + result = self.interface.home(axis_list=[0, 2]) + + self.assertEqual(result, SerialInterface.ReplyStatus.OK) + self.mock_serial_instance.send_command.assert_called_with("G28 A C\n", 10) + + def test_home_invalid_axis(self): + """Test homing with invalid axis index raises error.""" + self.interface.serial = self.mock_serial_instance + + with self.assertRaises(ValueError): + self.interface.home(axis_list=[10]) + + def test_calibrate_joint_no_save(self): + """Test calibrating a joint without saving results.""" + calibration_response = "0.5,1.0,100\n1.0,2.0,200\n1.5,3.0,300\n" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + calibration_response, + ) + self.interface.serial = self.mock_serial_instance + + result, data = self.interface.calibrate_joint(0, save_result=False) + + self.assertEqual(result, SerialInterface.ReplyStatus.OK) + self.assertEqual(len(data), 3) + self.assertEqual(data[0], [0.5, 1.0, 1.5]) # motor angles + self.assertEqual(data[1], [1.0, 2.0, 3.0]) # field angles + self.assertEqual(data[2], [100, 200, 300]) # encoder counts + self.mock_serial_instance.send_command.assert_called_with("M56 J0 P", 30) + + def test_calibrate_joint_with_save(self): + """Test calibrating a joint with saving results.""" + calibration_response = "0.5,1.0,100\n" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + calibration_response, + ) + self.interface.serial = self.mock_serial_instance + + result, data = self.interface.calibrate_joint(1, save_result=True) + + self.assertEqual(result, SerialInterface.ReplyStatus.OK) + self.mock_serial_instance.send_command.assert_called_with("M56 J1 P S", 30) + + def test_read_current_position(self): + """Test reading current position.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "X10.5 Y20.3 Z15.8", + ) + self.interface.serial = self.mock_serial_instance + + x, y, z = self.interface.read_current_position() + + self.assertAlmostEqual(x, 10.5) + self.assertAlmostEqual(y, 20.3) + self.assertAlmostEqual(z, 15.8) + self.mock_serial_instance.send_command.assert_called_with("M50") + + def test_read_current_position_error(self): + """Test reading current position returns None on error.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.ERROR, + "", + ) + self.interface.serial = self.mock_serial_instance + + x, y, z = self.interface.read_current_position() + + self.assertIsNone(x) + self.assertIsNone(y) + self.assertIsNone(z) + + def test_read_current_position_invalid_format(self): + """Test reading current position with invalid format raises error.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "invalid format", + ) + self.interface.serial = self.mock_serial_instance + + with self.assertRaises(ValueError): + self.interface.read_current_position() + + def test_move_to_immediate(self): + """Test moving to position with immediate execution.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "", + ) + self.interface.serial = self.mock_serial_instance + + result = self.interface.move_to(5.0, 10.0, 15.0, f=20.0, move_immediately=True) + + self.assertEqual(result, SerialInterface.ReplyStatus.OK) + call_args = self.mock_serial_instance.send_command.call_args + self.assertIn("G0 X5.000000 Y10.000000 Z15.000000 F20.000", call_args[0][0]) + self.assertIn("I", call_args[0][0]) + + def test_move_to_with_workspace_transform(self): + """Test move_to applies workspace transform correctly.""" + # Set a simple translation transform + transform = np.array([ + [1, 0, 0, 2], + [0, 1, 0, 3], + [0, 0, 1, 4], + [0, 0, 0, 1] + ]) + self.interface.set_workspace_transform(transform) + + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "", + ) + self.interface.serial = self.mock_serial_instance + + result = self.interface.move_to(0, 0, 0, f=10.0) + + # Expected transformed position is (2, 3, 4) + call_args = self.mock_serial_instance.send_command.call_args + cmd = call_args[0][0] + self.assertIn("X2.000000", cmd) + self.assertIn("Y3.000000", cmd) + self.assertIn("Z4.000000", cmd) + + def test_move_to_blocking_busy_retry(self): + """Test move_to retries on BUSY when blocking is True.""" + self.mock_serial_instance.send_command.side_effect = [ + (SerialInterface.ReplyStatus.BUSY, ""), + (SerialInterface.ReplyStatus.OK, ""), + ] + self.interface.serial = self.mock_serial_instance + + result = self.interface.move_to(5.0, 10.0, 15.0, f=20.0, blocking=True) + + self.assertEqual(result, SerialInterface.ReplyStatus.OK) + self.assertEqual(self.mock_serial_instance.send_command.call_count, 2) + + def test_move_to_non_blocking_returns_busy(self): + """Test move_to returns BUSY immediately when blocking is False.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.BUSY, + "", + ) + self.interface.serial = self.mock_serial_instance + + result = self.interface.move_to(5.0, 10.0, 15.0, f=20.0, blocking=False) + + self.assertEqual(result, SerialInterface.ReplyStatus.BUSY) + self.assertEqual(self.mock_serial_instance.send_command.call_count, 1) + + def test_dwell(self): + """Test dwell command.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "", + ) + self.interface.serial = self.mock_serial_instance + + result = self.interface.dwell(time_s=2.5, blocking=True) + + self.assertEqual(result, SerialInterface.ReplyStatus.OK) + call_args = self.mock_serial_instance.send_command.call_args + self.assertIn("G4 S2.500000", call_args[0][0]) + + def test_set_max_acceleration(self): + """Test setting max acceleration.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "", + ) + self.interface.serial = self.mock_serial_instance + + result = self.interface.set_max_acceleration( + linear_accel=100.0, angular_accel=50.0 + ) + + self.assertEqual(result, SerialInterface.ReplyStatus.OK) + call_args = self.mock_serial_instance.send_command.call_args + self.assertIn("M204 L100.000000 A50.000000", call_args[0][0]) + + def test_set_max_acceleration_minimum_values(self): + """Test max acceleration enforces minimum values.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "", + ) + self.interface.serial = self.mock_serial_instance + + result = self.interface.set_max_acceleration( + linear_accel=0.001, angular_accel=0.001 + ) + + call_args = self.mock_serial_instance.send_command.call_args + # Should be clamped to 0.01 + self.assertIn("M204 L0.010000 A0.010000", call_args[0][0]) + + def test_wait_for_stop_ready(self): + """Test wait_for_stop returns when device is ready.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "1", + ) + self.interface.serial = self.mock_serial_instance + + result = self.interface.wait_for_stop() + + self.assertEqual(result, SerialInterface.ReplyStatus.OK) + self.mock_serial_instance.send_command.assert_called_with("M53\n") + + def test_wait_for_stop_polls_until_ready(self): + """Test wait_for_stop polls until device is ready.""" + self.mock_serial_instance.send_command.side_effect = [ + (SerialInterface.ReplyStatus.OK, "0"), + (SerialInterface.ReplyStatus.OK, "0"), + (SerialInterface.ReplyStatus.OK, "1"), + ] + self.interface.serial = self.mock_serial_instance + + result = self.interface.wait_for_stop() + + self.assertEqual(result, SerialInterface.ReplyStatus.OK) + self.assertEqual(self.mock_serial_instance.send_command.call_count, 3) + + def test_wait_for_stop_error(self): + """Test wait_for_stop returns error status.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.ERROR, + "", + ) + self.interface.serial = self.mock_serial_instance + + result = self.interface.wait_for_stop() + + self.assertEqual(result, SerialInterface.ReplyStatus.ERROR) + + def test_enable_motors(self): + """Test enabling motors.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "", + ) + self.interface.serial = self.mock_serial_instance + + result = self.interface.enable_motors(enable=True) + + self.assertEqual(result, SerialInterface.ReplyStatus.OK) + self.mock_serial_instance.send_command.assert_called_with("M17", timeout=5) + + def test_disable_motors(self): + """Test disabling motors.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "", + ) + self.interface.serial = self.mock_serial_instance + + result = self.interface.enable_motors(enable=False) + + self.assertEqual(result, SerialInterface.ReplyStatus.OK) + self.mock_serial_instance.send_command.assert_called_with("M18", timeout=5) + + def test_set_pose(self): + """Test setting pose.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "", + ) + self.interface.serial = self.mock_serial_instance + + result = self.interface.set_pose(x=5.0, y=10.0, z=15.0) + + self.assertEqual(result, SerialInterface.ReplyStatus.OK) + call_args = self.mock_serial_instance.send_command.call_args + self.assertIn("G24 X5.000000 Y10.000000 Z15.000000", call_args[0][0]) + + def test_send_custom_command(self): + """Test sending custom command.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "response data", + ) + self.interface.serial = self.mock_serial_instance + + result, response = self.interface.send_command("M57", timeout_s=3.0) + + self.assertEqual(result, SerialInterface.ReplyStatus.OK) + self.assertEqual(response, "response data") + self.mock_serial_instance.send_command.assert_called_with("M57", 3.0) + + def test_read_device_state_info(self): + """Test reading device state info.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "state info", + ) + self.interface.serial = self.mock_serial_instance + + result = self.interface.read_device_state_info() + + self.assertEqual(result, SerialInterface.ReplyStatus.OK) + self.mock_serial_instance.send_command.assert_called_with("M57") + + def test_set_servo_parameter_defaults(self): + """Test setting servo parameters with defaults.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "", + ) + self.interface.serial = self.mock_serial_instance + + result = self.interface.set_servo_parameter() + + self.assertEqual(result, SerialInterface.ReplyStatus.OK) + call_args = self.mock_serial_instance.send_command.call_args + cmd = call_args[0][0] + self.assertIn("M55", cmd) + self.assertIn("A150.000000", cmd) # pos_kp + self.assertIn("B50000.000000", cmd) # pos_ki + self.assertIn("C0.200000", cmd) # vel_kp + self.assertIn("D100.000000", cmd) # vel_ki + self.assertIn("F0.002500", cmd) # vel_filter_tc + + def test_set_servo_parameter_custom(self): + """Test setting servo parameters with custom values.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "", + ) + self.interface.serial = self.mock_serial_instance + + result = self.interface.set_servo_parameter( + pos_kp=200, pos_ki=60000, vel_kp=0.3, vel_ki=120, vel_filter_tc=0.003 + ) + + self.assertEqual(result, SerialInterface.ReplyStatus.OK) + call_args = self.mock_serial_instance.send_command.call_args + cmd = call_args[0][0] + self.assertIn("A200.000000", cmd) + self.assertIn("B60000.000000", cmd) + self.assertIn("C0.300000", cmd) + self.assertIn("D120.000000", cmd) + self.assertIn("F0.003000", cmd) + + def test_read_encoder_angles(self): + """Test reading encoder angles returns empty list.""" + self.mock_serial_instance.send_command.return_value = ( + SerialInterface.ReplyStatus.OK, + "", + ) + self.interface.serial = self.mock_serial_instance + + result = self.interface.read_encoder_angles() + + self.assertEqual(result, []) + self.mock_serial_instance.send_command.assert_called_with("M51") + + def test_parse_table_data(self): + """Test parsing table data.""" + data_string = "1.0,2.0,3.0\n4.0,5.0,6.0\n7.0,8.0,9.0" + result = OpenMicroStageInterface._parse_table_data(data_string, 3) + + self.assertEqual(len(result), 3) + self.assertEqual(result[0], [1.0, 4.0, 7.0]) + self.assertEqual(result[1], [2.0, 5.0, 8.0]) + self.assertEqual(result[2], [3.0, 6.0, 9.0]) + + def test_parse_table_data_with_malformed_lines(self): + """Test parsing table data skips malformed lines.""" + data_string = "1.0,2.0,3.0\ninvalid\n4.0,5.0,6.0" + result = OpenMicroStageInterface._parse_table_data(data_string, 3) + + self.assertEqual(len(result), 3) + self.assertEqual(result[0], [1.0, 4.0]) + self.assertEqual(result[1], [2.0, 5.0]) + self.assertEqual(result[2], [3.0, 6.0]) + + def test_parse_table_data_single_row(self): + """Test parsing single row of data.""" + data_string = "10.5,20.3,15.8" + result = OpenMicroStageInterface._parse_table_data(data_string, 3) + + self.assertEqual(result[0], [10.5]) + self.assertEqual(result[1], [20.3]) + self.assertEqual(result[2], [15.8]) + + def test_workflow_connect_home_move_stop(self): + """Test end-to-end workflow: connect, home, move, wait for stop.""" + # Mock return values for each command + self.mock_serial_instance.send_command.side_effect = [ + (SerialInterface.ReplyStatus.OK, "v1.0.1"), # firmware version + (SerialInterface.ReplyStatus.OK, ""), # home + (SerialInterface.ReplyStatus.OK, ""), # move_to + (SerialInterface.ReplyStatus.OK, "1"), # wait_for_stop + ] + + self.interface.connect("/dev/ttyACM0") + self.assertIsNotNone(self.interface.serial) + + home_result = self.interface.home() + self.assertEqual(home_result, SerialInterface.ReplyStatus.OK) + + move_result = self.interface.move_to(5.0, 10.0, 15.0, f=20.0) + self.assertEqual(move_result, SerialInterface.ReplyStatus.OK) + + stop_result = self.interface.wait_for_stop() + self.assertEqual(stop_result, SerialInterface.ReplyStatus.OK) + + def test_workflow_calibrate_and_move(self): + """Test end-to-end workflow: calibrate joint and then move.""" + calibration_data = "0.5,1.0,100\n1.0,2.0,200\n" + + self.mock_serial_instance.send_command.side_effect = [ + (SerialInterface.ReplyStatus.OK, calibration_data), # calibrate + (SerialInterface.ReplyStatus.OK, "X5.0 Y10.0 Z15.0"), # read position + ] + + self.interface.serial = self.mock_serial_instance + + result, data = self.interface.calibrate_joint(0, save_result=True) + self.assertEqual(result, SerialInterface.ReplyStatus.OK) + self.assertEqual(len(data), 3) + + x, y, z = self.interface.read_current_position() + self.assertAlmostEqual(x, 5.0) + self.assertAlmostEqual(y, 10.0) + self.assertAlmostEqual(z, 15.0) + + +if __name__ == "__main__": + unittest.main() From 50eb6e9969ac7f3b4ae4934552e82bd2bbfd6752 Mon Sep 17 00:00:00 2001 From: Michael Honaker Date: Mon, 9 Feb 2026 15:20:53 -0500 Subject: [PATCH 05/14] Continued updates --- software/PythonAPI/tests/conftest.py | 2 + .../PythonAPI/tests/test_open_micro_stage.py | 299 ++++++++++-------- 2 files changed, 172 insertions(+), 129 deletions(-) create mode 100644 software/PythonAPI/tests/conftest.py diff --git a/software/PythonAPI/tests/conftest.py b/software/PythonAPI/tests/conftest.py new file mode 100644 index 0000000..83e3114 --- /dev/null +++ b/software/PythonAPI/tests/conftest.py @@ -0,0 +1,2 @@ +"""Shared test fixtures and utilities.""" + diff --git a/software/PythonAPI/tests/test_open_micro_stage.py b/software/PythonAPI/tests/test_open_micro_stage.py index a3d52fd..9d15fce 100644 --- a/software/PythonAPI/tests/test_open_micro_stage.py +++ b/software/PythonAPI/tests/test_open_micro_stage.py @@ -2,31 +2,122 @@ import re import unittest -from unittest.mock import MagicMock, Mock, patch +from unittest.mock import patch import numpy as np from open_micro_stage.api import OpenMicroStageInterface, SerialInterface +class MockSerialInterface: + """Generic mock implementation of SerialInterface for testing.""" + + def __init__( + self, + port: str = "/dev/ttyACM0", + baud_rate: int = 115200, + command_msg_callback=None, + log_msg_callback=None, + unsolicited_msg_callback=None, + reconnect_timeout: int = 5, + ): + """Initialize mock serial interface.""" + self.port = port + self.baud_rate = baud_rate + self.reconnect_timeout = reconnect_timeout + self.command_msg_callback = command_msg_callback + self.log_message_callback = log_msg_callback + self.unsolicited_msg_callback = unsolicited_msg_callback + + # Command response queue + self.responses = [] + self.response_index = 0 + self.call_history = [] + + def set_response(self, status: SerialInterface.ReplyStatus, response: str): + """Set a single response for the next send_command call.""" + self.responses = [(status, response)] + self.response_index = 0 + + def set_responses(self, responses: list): + """Set multiple responses for sequential send_command calls. + + Args: + responses: List of (status, response) tuples + """ + self.responses = responses + self.response_index = 0 + + def send_command(self, cmd: str, timeout=2): + """Send a command and return a mocked response.""" + self.call_history.append((cmd, timeout)) + + if self.response_index >= len(self.responses): + # If we run out of responses, return the last one or OK + if self.responses: + status, response = self.responses[-1] + else: + status, response = SerialInterface.ReplyStatus.OK, "" + else: + status, response = self.responses[self.response_index] + self.response_index += 1 + + # Call the command callback if set + if self.command_msg_callback: + self.command_msg_callback(cmd, None, "") + + return status, response + + def close(self): + """Close the mock connection.""" + pass + + def reset(self): + """Reset mock state for a new test.""" + self.responses = [] + self.response_index = 0 + self.call_history = [] + + def get_last_command(self): + """Get the last command that was sent.""" + if self.call_history: + return self.call_history[-1][0] + return None + + def get_all_commands(self): + """Get all commands that were sent.""" + return [cmd for cmd, _ in self.call_history] + + def assert_command_called(self, cmd: str): + """Assert that a specific command was called.""" + if cmd not in self.get_all_commands(): + raise AssertionError(f"Command '{cmd}' was not called. Commands: {self.get_all_commands()}") + + def assert_command_called_with_args(self, cmd: str, timeout=None): + """Assert that a specific command was called with specific arguments.""" + for called_cmd, called_timeout in self.call_history: + if called_cmd == cmd and (timeout is None or called_timeout == timeout): + return + raise AssertionError( + f"Command '{cmd}' with timeout={timeout} was not called. " + f"Call history: {self.call_history}" + ) + + class TestOpenMicroStageInterface(unittest.TestCase): """End-to-end tests for OpenMicroStageInterface.""" def setUp(self): """Set up test fixtures with mocked SerialInterface.""" - # Patch SerialInterface.__init__ to prevent actual connection attempts - self.patcher = patch("open_micro_stage.api.SerialInterface.__init__", return_value=None) - self.mock_init = self.patcher.start() - - # Create a mock serial instance that we can control - self.mock_serial_instance = MagicMock(spec=SerialInterface) + # Create a persistent mock instance + self.mock_serial_instance = MockSerialInterface() - # Patch the SerialInterface class to return our mock instance - self.class_patcher = patch( + # Patch SerialInterface to return the same mock instance every time + self.patcher = patch( "open_micro_stage.api.SerialInterface", return_value=self.mock_serial_instance ) - self.mock_serial_class = self.class_patcher.start() + self.patcher.start() # Create the interface self.interface = OpenMicroStageInterface( @@ -36,7 +127,6 @@ def setUp(self): def tearDown(self): """Clean up patches.""" self.patcher.stop() - self.class_patcher.stop() def test_initialization(self): """Test that OpenMicroStageInterface initializes with correct defaults.""" @@ -57,42 +147,29 @@ def test_initialization_with_params(self): def test_connect_success(self): """Test successful connection to device.""" - self.mock_serial_instance.send_command.return_value = ( + self.mock_serial_instance.set_response( SerialInterface.ReplyStatus.OK, "v1.0.1", ) - self.interface.connect("/dev/ttyACM0") self.assertIsNotNone(self.interface.serial) - self.mock_serial_class.assert_called_once_with( - "/dev/ttyACM0", - 921600, - log_msg_callback=self.interface.log_msg_callback, - command_msg_callback=self.interface.command_msg_callback, - unsolicited_msg_callback=self.interface.unsolicited_msg_callback, - ) def test_connect_with_custom_baud_rate(self): """Test connection with custom baud rate.""" - self.mock_serial_instance.send_command.return_value = ( + self.mock_serial_instance.set_response( SerialInterface.ReplyStatus.OK, "v1.0.1", ) self.interface.connect("/dev/ttyACM0", baud_rate=115200) - self.mock_serial_class.assert_called_once_with( - "/dev/ttyACM0", - 115200, - log_msg_callback=self.interface.log_msg_callback, - command_msg_callback=self.interface.command_msg_callback, - unsolicited_msg_callback=self.interface.unsolicited_msg_callback, - ) + # Verify the mock was called with correct parameters + self.assertIsNotNone(self.interface.serial) def test_connect_incompatible_firmware(self): """Test connection fails with incompatible firmware version.""" - self.mock_serial_instance.send_command.return_value = ( + self.mock_serial_instance.set_response( SerialInterface.ReplyStatus.OK, "v0.9.0", ) @@ -104,7 +181,7 @@ def test_connect_incompatible_firmware(self): def test_disconnect(self): """Test disconnection from device.""" - self.mock_serial_instance.send_command.return_value = ( + self.mock_serial_instance.set_response( SerialInterface.ReplyStatus.OK, "v1.0.1", ) @@ -112,7 +189,6 @@ def test_disconnect(self): self.interface.connect("/dev/ttyACM0") self.interface.disconnect() - self.mock_serial_instance.close.assert_called_once() self.assertIsNone(self.interface.serial) def test_disconnect_when_not_connected(self): @@ -137,26 +213,24 @@ def test_set_and_get_workspace_transform(self): def test_read_firmware_version(self): """Test reading firmware version.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "v1.2.3", ) - self.interface.serial = self.mock_serial_instance major, minor, patch = self.interface.read_firmware_version() self.assertEqual(major, 1) self.assertEqual(minor, 2) self.assertEqual(patch, 3) - self.mock_serial_instance.send_command.assert_called_with("M58") + self.interface.serial.assert_command_called("M58") def test_read_firmware_version_error(self): """Test firmware version returns 0,0,0 on error.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.ERROR, "", ) - self.interface.serial = self.mock_serial_instance major, minor, patch = self.interface.read_firmware_version() @@ -164,45 +238,40 @@ def test_read_firmware_version_error(self): def test_home_all_axes(self): """Test homing all axes.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "", ) - self.interface.serial = self.mock_serial_instance result = self.interface.home() self.assertEqual(result, SerialInterface.ReplyStatus.OK) - self.mock_serial_instance.send_command.assert_called_with("G28 A B C D E F\n", 10) + self.interface.serial.assert_command_called("G28 A B C D E F\n") def test_home_specific_axes(self): """Test homing specific axes.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "", ) - self.interface.serial = self.mock_serial_instance result = self.interface.home(axis_list=[0, 2]) self.assertEqual(result, SerialInterface.ReplyStatus.OK) - self.mock_serial_instance.send_command.assert_called_with("G28 A C\n", 10) + self.interface.serial.assert_command_called("G28 A C\n") def test_home_invalid_axis(self): """Test homing with invalid axis index raises error.""" - self.interface.serial = self.mock_serial_instance - with self.assertRaises(ValueError): self.interface.home(axis_list=[10]) def test_calibrate_joint_no_save(self): """Test calibrating a joint without saving results.""" calibration_response = "0.5,1.0,100\n1.0,2.0,200\n1.5,3.0,300\n" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, calibration_response, ) - self.interface.serial = self.mock_serial_instance result, data = self.interface.calibrate_joint(0, save_result=False) @@ -211,44 +280,41 @@ def test_calibrate_joint_no_save(self): self.assertEqual(data[0], [0.5, 1.0, 1.5]) # motor angles self.assertEqual(data[1], [1.0, 2.0, 3.0]) # field angles self.assertEqual(data[2], [100, 200, 300]) # encoder counts - self.mock_serial_instance.send_command.assert_called_with("M56 J0 P", 30) + self.interface.serial.assert_command_called("M56 J0 P") def test_calibrate_joint_with_save(self): """Test calibrating a joint with saving results.""" calibration_response = "0.5,1.0,100\n" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, calibration_response, ) - self.interface.serial = self.mock_serial_instance result, data = self.interface.calibrate_joint(1, save_result=True) self.assertEqual(result, SerialInterface.ReplyStatus.OK) - self.mock_serial_instance.send_command.assert_called_with("M56 J1 P S", 30) + self.interface.serial.assert_command_called("M56 J1 P S") def test_read_current_position(self): """Test reading current position.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "X10.5 Y20.3 Z15.8", ) - self.interface.serial = self.mock_serial_instance x, y, z = self.interface.read_current_position() self.assertAlmostEqual(x, 10.5) self.assertAlmostEqual(y, 20.3) self.assertAlmostEqual(z, 15.8) - self.mock_serial_instance.send_command.assert_called_with("M50") + self.interface.serial.assert_command_called("M50") def test_read_current_position_error(self): """Test reading current position returns None on error.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.ERROR, "", ) - self.interface.serial = self.mock_serial_instance x, y, z = self.interface.read_current_position() @@ -258,29 +324,27 @@ def test_read_current_position_error(self): def test_read_current_position_invalid_format(self): """Test reading current position with invalid format raises error.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "invalid format", ) - self.interface.serial = self.mock_serial_instance with self.assertRaises(ValueError): self.interface.read_current_position() def test_move_to_immediate(self): """Test moving to position with immediate execution.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "", ) - self.interface.serial = self.mock_serial_instance result = self.interface.move_to(5.0, 10.0, 15.0, f=20.0, move_immediately=True) self.assertEqual(result, SerialInterface.ReplyStatus.OK) - call_args = self.mock_serial_instance.send_command.call_args - self.assertIn("G0 X5.000000 Y10.000000 Z15.000000 F20.000", call_args[0][0]) - self.assertIn("I", call_args[0][0]) + cmd = self.interface.serial.get_last_command() + self.assertIn("G0 X5.000000 Y10.000000 Z15.000000 F20.000", cmd) + self.assertIn("I", cmd) def test_move_to_with_workspace_transform(self): """Test move_to applies workspace transform correctly.""" @@ -293,127 +357,117 @@ def test_move_to_with_workspace_transform(self): ]) self.interface.set_workspace_transform(transform) - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "", ) - self.interface.serial = self.mock_serial_instance result = self.interface.move_to(0, 0, 0, f=10.0) # Expected transformed position is (2, 3, 4) - call_args = self.mock_serial_instance.send_command.call_args - cmd = call_args[0][0] + cmd = self.interface.serial.get_last_command() self.assertIn("X2.000000", cmd) self.assertIn("Y3.000000", cmd) self.assertIn("Z4.000000", cmd) def test_move_to_blocking_busy_retry(self): """Test move_to retries on BUSY when blocking is True.""" - self.mock_serial_instance.send_command.side_effect = [ + self.interface.serial.set_responses([ (SerialInterface.ReplyStatus.BUSY, ""), (SerialInterface.ReplyStatus.OK, ""), - ] - self.interface.serial = self.mock_serial_instance + ]) - result = self.interface.move_to(5.0, 10.0, 15.0, f=20.0, blocking=True) + result = self.interface.move_to(5.0, 10.0, 15.0, f=20.0, blocking=True, timeout=0.01) self.assertEqual(result, SerialInterface.ReplyStatus.OK) - self.assertEqual(self.mock_serial_instance.send_command.call_count, 2) + self.assertEqual(len(self.interface.serial.call_history), 2) def test_move_to_non_blocking_returns_busy(self): """Test move_to returns BUSY immediately when blocking is False.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.BUSY, "", ) - self.interface.serial = self.mock_serial_instance result = self.interface.move_to(5.0, 10.0, 15.0, f=20.0, blocking=False) self.assertEqual(result, SerialInterface.ReplyStatus.BUSY) - self.assertEqual(self.mock_serial_instance.send_command.call_count, 1) + self.assertEqual(len(self.interface.serial.call_history), 1) def test_dwell(self): """Test dwell command.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "", ) - self.interface.serial = self.mock_serial_instance result = self.interface.dwell(time_s=2.5, blocking=True) self.assertEqual(result, SerialInterface.ReplyStatus.OK) - call_args = self.mock_serial_instance.send_command.call_args - self.assertIn("G4 S2.500000", call_args[0][0]) + cmd = self.interface.serial.get_last_command() + self.assertIn("G4 S2.500000", cmd) def test_set_max_acceleration(self): """Test setting max acceleration.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "", ) - self.interface.serial = self.mock_serial_instance result = self.interface.set_max_acceleration( linear_accel=100.0, angular_accel=50.0 ) self.assertEqual(result, SerialInterface.ReplyStatus.OK) - call_args = self.mock_serial_instance.send_command.call_args - self.assertIn("M204 L100.000000 A50.000000", call_args[0][0]) + cmd = self.interface.serial.get_last_command() + self.assertIn("M204 L100.000000 A50.000000", cmd) def test_set_max_acceleration_minimum_values(self): """Test max acceleration enforces minimum values.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "", ) - self.interface.serial = self.mock_serial_instance result = self.interface.set_max_acceleration( linear_accel=0.001, angular_accel=0.001 ) - call_args = self.mock_serial_instance.send_command.call_args + cmd = self.interface.serial.get_last_command() # Should be clamped to 0.01 - self.assertIn("M204 L0.010000 A0.010000", call_args[0][0]) + self.assertIn("M204 L0.010000 A0.010000", cmd) def test_wait_for_stop_ready(self): """Test wait_for_stop returns when device is ready.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "1", ) - self.interface.serial = self.mock_serial_instance result = self.interface.wait_for_stop() self.assertEqual(result, SerialInterface.ReplyStatus.OK) - self.mock_serial_instance.send_command.assert_called_with("M53\n") + self.interface.serial.assert_command_called("M53\n") def test_wait_for_stop_polls_until_ready(self): """Test wait_for_stop polls until device is ready.""" - self.mock_serial_instance.send_command.side_effect = [ + self.interface.serial.set_responses([ (SerialInterface.ReplyStatus.OK, "0"), (SerialInterface.ReplyStatus.OK, "0"), (SerialInterface.ReplyStatus.OK, "1"), - ] - self.interface.serial = self.mock_serial_instance + ]) result = self.interface.wait_for_stop() self.assertEqual(result, SerialInterface.ReplyStatus.OK) - self.assertEqual(self.mock_serial_instance.send_command.call_count, 3) + self.assertEqual(len(self.interface.serial.call_history), 3) def test_wait_for_stop_error(self): """Test wait_for_stop returns error status.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.ERROR, "", ) - self.interface.serial = self.mock_serial_instance result = self.interface.wait_for_stop() @@ -421,84 +475,76 @@ def test_wait_for_stop_error(self): def test_enable_motors(self): """Test enabling motors.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "", ) - self.interface.serial = self.mock_serial_instance result = self.interface.enable_motors(enable=True) self.assertEqual(result, SerialInterface.ReplyStatus.OK) - self.mock_serial_instance.send_command.assert_called_with("M17", timeout=5) + self.interface.serial.assert_command_called_with_args("M17", timeout=5) def test_disable_motors(self): """Test disabling motors.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "", ) - self.interface.serial = self.mock_serial_instance result = self.interface.enable_motors(enable=False) self.assertEqual(result, SerialInterface.ReplyStatus.OK) - self.mock_serial_instance.send_command.assert_called_with("M18", timeout=5) + self.interface.serial.assert_command_called_with_args("M18", timeout=5) def test_set_pose(self): """Test setting pose.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "", ) - self.interface.serial = self.mock_serial_instance result = self.interface.set_pose(x=5.0, y=10.0, z=15.0) self.assertEqual(result, SerialInterface.ReplyStatus.OK) - call_args = self.mock_serial_instance.send_command.call_args - self.assertIn("G24 X5.000000 Y10.000000 Z15.000000", call_args[0][0]) + cmd = self.interface.serial.get_last_command() + self.assertIn("G24 X5.000000 Y10.000000 Z15.000000", cmd) def test_send_custom_command(self): """Test sending custom command.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "response data", ) - self.interface.serial = self.mock_serial_instance result, response = self.interface.send_command("M57", timeout_s=3.0) self.assertEqual(result, SerialInterface.ReplyStatus.OK) self.assertEqual(response, "response data") - self.mock_serial_instance.send_command.assert_called_with("M57", 3.0) def test_read_device_state_info(self): """Test reading device state info.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "state info", ) - self.interface.serial = self.mock_serial_instance result = self.interface.read_device_state_info() self.assertEqual(result, SerialInterface.ReplyStatus.OK) - self.mock_serial_instance.send_command.assert_called_with("M57") + self.interface.serial.assert_command_called("M57") def test_set_servo_parameter_defaults(self): """Test setting servo parameters with defaults.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "", ) - self.interface.serial = self.mock_serial_instance result = self.interface.set_servo_parameter() self.assertEqual(result, SerialInterface.ReplyStatus.OK) - call_args = self.mock_serial_instance.send_command.call_args - cmd = call_args[0][0] + cmd = self.interface.serial.get_last_command() self.assertIn("M55", cmd) self.assertIn("A150.000000", cmd) # pos_kp self.assertIn("B50000.000000", cmd) # pos_ki @@ -508,19 +554,17 @@ def test_set_servo_parameter_defaults(self): def test_set_servo_parameter_custom(self): """Test setting servo parameters with custom values.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "", ) - self.interface.serial = self.mock_serial_instance result = self.interface.set_servo_parameter( pos_kp=200, pos_ki=60000, vel_kp=0.3, vel_ki=120, vel_filter_tc=0.003 ) self.assertEqual(result, SerialInterface.ReplyStatus.OK) - call_args = self.mock_serial_instance.send_command.call_args - cmd = call_args[0][0] + cmd = self.interface.serial.get_last_command() self.assertIn("A200.000000", cmd) self.assertIn("B60000.000000", cmd) self.assertIn("C0.300000", cmd) @@ -529,16 +573,15 @@ def test_set_servo_parameter_custom(self): def test_read_encoder_angles(self): """Test reading encoder angles returns empty list.""" - self.mock_serial_instance.send_command.return_value = ( + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "", ) - self.interface.serial = self.mock_serial_instance result = self.interface.read_encoder_angles() self.assertEqual(result, []) - self.mock_serial_instance.send_command.assert_called_with("M51") + self.interface.serial.assert_command_called("M51") def test_parse_table_data(self): """Test parsing table data.""" @@ -571,13 +614,13 @@ def test_parse_table_data_single_row(self): def test_workflow_connect_home_move_stop(self): """Test end-to-end workflow: connect, home, move, wait for stop.""" - # Mock return values for each command - self.mock_serial_instance.send_command.side_effect = [ + # Set responses for each command in sequence + self.mock_serial_instance.set_responses([ (SerialInterface.ReplyStatus.OK, "v1.0.1"), # firmware version (SerialInterface.ReplyStatus.OK, ""), # home (SerialInterface.ReplyStatus.OK, ""), # move_to (SerialInterface.ReplyStatus.OK, "1"), # wait_for_stop - ] + ]) self.interface.connect("/dev/ttyACM0") self.assertIsNotNone(self.interface.serial) @@ -595,12 +638,10 @@ def test_workflow_calibrate_and_move(self): """Test end-to-end workflow: calibrate joint and then move.""" calibration_data = "0.5,1.0,100\n1.0,2.0,200\n" - self.mock_serial_instance.send_command.side_effect = [ + self.mock_serial_instance.set_responses([ (SerialInterface.ReplyStatus.OK, calibration_data), # calibrate (SerialInterface.ReplyStatus.OK, "X5.0 Y10.0 Z15.0"), # read position - ] - - self.interface.serial = self.mock_serial_instance + ]) result, data = self.interface.calibrate_joint(0, save_result=True) self.assertEqual(result, SerialInterface.ReplyStatus.OK) From f21e56a749e85223d5bef6df2f938a9d2f85b179 Mon Sep 17 00:00:00 2001 From: Michael Honaker Date: Mon, 9 Feb 2026 15:29:04 -0500 Subject: [PATCH 06/14] Alot working --- software/PythonAPI/tests/conftest.py | 2 -- software/PythonAPI/tests/test_open_micro_stage.py | 11 ++++++++++- 2 files changed, 10 insertions(+), 3 deletions(-) delete mode 100644 software/PythonAPI/tests/conftest.py diff --git a/software/PythonAPI/tests/conftest.py b/software/PythonAPI/tests/conftest.py deleted file mode 100644 index 83e3114..0000000 --- a/software/PythonAPI/tests/conftest.py +++ /dev/null @@ -1,2 +0,0 @@ -"""Shared test fixtures and utilities.""" - diff --git a/software/PythonAPI/tests/test_open_micro_stage.py b/software/PythonAPI/tests/test_open_micro_stage.py index 9d15fce..2b29f27 100644 --- a/software/PythonAPI/tests/test_open_micro_stage.py +++ b/software/PythonAPI/tests/test_open_micro_stage.py @@ -11,6 +11,8 @@ class MockSerialInterface: """Generic mock implementation of SerialInterface for testing.""" + ReplyStatus = SerialInterface.ReplyStatus + LogLevel = SerialInterface.LogLevel def __init__( self, @@ -114,7 +116,7 @@ def setUp(self): # Patch SerialInterface to return the same mock instance every time self.patcher = patch( - "open_micro_stage.api.SerialInterface", + "open_micro_stage.api.SerialInterface.__new__", return_value=self.mock_serial_instance ) self.patcher.start() @@ -124,6 +126,13 @@ def setUp(self): show_communication=False, show_log_messages=False ) + # default connect for the majority of commands + self.mock_serial_instance.set_response( + SerialInterface.ReplyStatus.OK, + "v1.0.1", + ) + self.interface.connect("/dev/ttyACM0") + def tearDown(self): """Clean up patches.""" self.patcher.stop() From 038ae01ab902f999f08e924ccb55147acbb3f610 Mon Sep 17 00:00:00 2001 From: Michael Honaker Date: Mon, 9 Feb 2026 15:34:15 -0500 Subject: [PATCH 07/14] Updates with fixed tests --- software/PythonAPI/open_micro_stage/api.py | 2 +- software/PythonAPI/tests/test_open_micro_stage.py | 7 ++++--- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/software/PythonAPI/open_micro_stage/api.py b/software/PythonAPI/open_micro_stage/api.py index 7da6bc3..02b77fb 100644 --- a/software/PythonAPI/open_micro_stage/api.py +++ b/software/PythonAPI/open_micro_stage/api.py @@ -314,7 +314,7 @@ def home(self, axis_list=None): axis_list = list(range(len(axis_chars))) for axis_idx in axis_list: - if 0 > axis_idx >= len(axis_chars): + if (0 > axis_idx) or (axis_idx >= len(axis_chars)): raise ValueError("Axis index out of range") cmd += " " + axis_chars[axis_idx] diff --git a/software/PythonAPI/tests/test_open_micro_stage.py b/software/PythonAPI/tests/test_open_micro_stage.py index 2b29f27..6ea49b8 100644 --- a/software/PythonAPI/tests/test_open_micro_stage.py +++ b/software/PythonAPI/tests/test_open_micro_stage.py @@ -132,6 +132,7 @@ def setUp(self): "v1.0.1", ) self.interface.connect("/dev/ttyACM0") + self.calls_for_initailization = len(self.mock_serial_instance.call_history) def tearDown(self): """Clean up patches.""" @@ -389,7 +390,7 @@ def test_move_to_blocking_busy_retry(self): result = self.interface.move_to(5.0, 10.0, 15.0, f=20.0, blocking=True, timeout=0.01) self.assertEqual(result, SerialInterface.ReplyStatus.OK) - self.assertEqual(len(self.interface.serial.call_history), 2) + self.assertEqual(len(self.interface.serial.call_history), 2+self.calls_for_initailization) def test_move_to_non_blocking_returns_busy(self): """Test move_to returns BUSY immediately when blocking is False.""" @@ -401,7 +402,7 @@ def test_move_to_non_blocking_returns_busy(self): result = self.interface.move_to(5.0, 10.0, 15.0, f=20.0, blocking=False) self.assertEqual(result, SerialInterface.ReplyStatus.BUSY) - self.assertEqual(len(self.interface.serial.call_history), 1) + self.assertEqual(len(self.interface.serial.call_history), 1+self.calls_for_initailization) def test_dwell(self): """Test dwell command.""" @@ -469,7 +470,7 @@ def test_wait_for_stop_polls_until_ready(self): result = self.interface.wait_for_stop() self.assertEqual(result, SerialInterface.ReplyStatus.OK) - self.assertEqual(len(self.interface.serial.call_history), 3) + self.assertEqual(len(self.interface.serial.call_history), 3+self.calls_for_initailization) def test_wait_for_stop_error(self): """Test wait_for_stop returns error status.""" From 35278b92f293abaf6781d0f444b7cb7774efba8e Mon Sep 17 00:00:00 2001 From: Michael Honaker Date: Mon, 9 Feb 2026 15:36:13 -0500 Subject: [PATCH 08/14] Minor test updates --- software/PythonAPI/tests/test_open_micro_stage.py | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/software/PythonAPI/tests/test_open_micro_stage.py b/software/PythonAPI/tests/test_open_micro_stage.py index 6ea49b8..21ba886 100644 --- a/software/PythonAPI/tests/test_open_micro_stage.py +++ b/software/PythonAPI/tests/test_open_micro_stage.py @@ -8,6 +8,7 @@ from open_micro_stage.api import OpenMicroStageInterface, SerialInterface +DESIRED_FIRMWARE_VERSION = "v1.0.1" class MockSerialInterface: """Generic mock implementation of SerialInterface for testing.""" @@ -129,7 +130,7 @@ def setUp(self): # default connect for the majority of commands self.mock_serial_instance.set_response( SerialInterface.ReplyStatus.OK, - "v1.0.1", + DESIRED_FIRMWARE_VERSION, ) self.interface.connect("/dev/ttyACM0") self.calls_for_initailization = len(self.mock_serial_instance.call_history) @@ -159,23 +160,25 @@ def test_connect_success(self): """Test successful connection to device.""" self.mock_serial_instance.set_response( SerialInterface.ReplyStatus.OK, - "v1.0.1", + DESIRED_FIRMWARE_VERSION, ) self.interface.connect("/dev/ttyACM0") self.assertIsNotNone(self.interface.serial) + self.assertEqual(self.interface.serial.port, "/dev/ttyACM0") def test_connect_with_custom_baud_rate(self): """Test connection with custom baud rate.""" self.mock_serial_instance.set_response( SerialInterface.ReplyStatus.OK, - "v1.0.1", + DESIRED_FIRMWARE_VERSION, ) self.interface.connect("/dev/ttyACM0", baud_rate=115200) # Verify the mock was called with correct parameters self.assertIsNotNone(self.interface.serial) + self.assertEqual(self.interface.serial.baud_rate, 115200) def test_connect_incompatible_firmware(self): """Test connection fails with incompatible firmware version.""" @@ -193,7 +196,7 @@ def test_disconnect(self): """Test disconnection from device.""" self.mock_serial_instance.set_response( SerialInterface.ReplyStatus.OK, - "v1.0.1", + DESIRED_FIRMWARE_VERSION, ) self.interface.connect("/dev/ttyACM0") @@ -626,7 +629,7 @@ def test_workflow_connect_home_move_stop(self): """Test end-to-end workflow: connect, home, move, wait for stop.""" # Set responses for each command in sequence self.mock_serial_instance.set_responses([ - (SerialInterface.ReplyStatus.OK, "v1.0.1"), # firmware version + (SerialInterface.ReplyStatus.OK, DESIRED_FIRMWARE_VERSION), # firmware version (SerialInterface.ReplyStatus.OK, ""), # home (SerialInterface.ReplyStatus.OK, ""), # move_to (SerialInterface.ReplyStatus.OK, "1"), # wait_for_stop From feed1fc03883b960df429124d7d609156f385c52 Mon Sep 17 00:00:00 2001 From: Michael Honaker Date: Mon, 9 Feb 2026 15:51:50 -0500 Subject: [PATCH 09/14] Update with changes --- software/PythonAPI/.gitignore | 1 + software/PythonAPI/open_micro_stage/utils.py | 24 +- .../PythonAPI/tests/test_open_micro_stage.py | 244 +++++++++--------- 3 files changed, 130 insertions(+), 139 deletions(-) diff --git a/software/PythonAPI/.gitignore b/software/PythonAPI/.gitignore index f4daebb..297dd9a 100644 --- a/software/PythonAPI/.gitignore +++ b/software/PythonAPI/.gitignore @@ -26,6 +26,7 @@ share/python-wheels/ .installed.cfg *.egg MANIFEST +.ruff_cache/ # PyInstaller *.manifest diff --git a/software/PythonAPI/open_micro_stage/utils.py b/software/PythonAPI/open_micro_stage/utils.py index a555a1f..57bcabe 100644 --- a/software/PythonAPI/open_micro_stage/utils.py +++ b/software/PythonAPI/open_micro_stage/utils.py @@ -1,34 +1,36 @@ - type LocalArray = list[float] type LocalMatrix = list[LocalArray] -def generate_eye(size: int)->LocalMatrix: + +def generate_eye(size: int) -> LocalMatrix: """Generate a 2D list representing an eye pattern.""" return [[1 if i == j else 0 for j in range(size)] for i in range(size)] -def matrix_multiply(A: LocalMatrix, B: LocalMatrix)->LocalMatrix: + +def matrix_multiply(A: LocalMatrix, B: LocalMatrix) -> LocalMatrix: """Multiply two matrices A and B.""" if len(A[0]) != len(B): raise ValueError("Number of columns in A must equal number of rows in B.") - + result = [[0 for _ in range(len(B[0]))] for _ in range(len(A))] - + for i in range(len(A)): for j in range(len(B[0])): for k in range(len(B)): result[i][j] += A[i][k] * B[k][j] - + return result -def matrix_dot_product(A: LocalMatrix, B: LocalArray)->LocalArray: + +def matrix_dot_product(A: LocalMatrix, B: LocalArray) -> LocalArray: """Calculate the dot product of two matrices A and B.""" if len(A[0]) != len(B): raise ValueError("Number of columns in A must equal number of elements in B.") - + result = [0 for _ in range(len(A))] - + for i in range(len(A)): for j in range(len(B)): result[i] += A[i][j] * B[j] - - return result \ No newline at end of file + + return result diff --git a/software/PythonAPI/tests/test_open_micro_stage.py b/software/PythonAPI/tests/test_open_micro_stage.py index 21ba886..e502275 100644 --- a/software/PythonAPI/tests/test_open_micro_stage.py +++ b/software/PythonAPI/tests/test_open_micro_stage.py @@ -1,6 +1,5 @@ """Tests for the OpenMicroStageInterface class.""" -import re import unittest from unittest.mock import patch @@ -10,8 +9,10 @@ DESIRED_FIRMWARE_VERSION = "v1.0.1" + class MockSerialInterface: """Generic mock implementation of SerialInterface for testing.""" + ReplyStatus = SerialInterface.ReplyStatus LogLevel = SerialInterface.LogLevel @@ -102,8 +103,7 @@ def assert_command_called_with_args(self, cmd: str, timeout=None): if called_cmd == cmd and (timeout is None or called_timeout == timeout): return raise AssertionError( - f"Command '{cmd}' with timeout={timeout} was not called. " - f"Call history: {self.call_history}" + f"Command '{cmd}' with timeout={timeout} was not called. Call history: {self.call_history}" ) @@ -114,18 +114,13 @@ def setUp(self): """Set up test fixtures with mocked SerialInterface.""" # Create a persistent mock instance self.mock_serial_instance = MockSerialInterface() - + # Patch SerialInterface to return the same mock instance every time - self.patcher = patch( - "open_micro_stage.api.SerialInterface.__new__", - return_value=self.mock_serial_instance - ) + self.patcher = patch("open_micro_stage.api.SerialInterface.__new__", return_value=self.mock_serial_instance) self.patcher.start() - + # Create the interface - self.interface = OpenMicroStageInterface( - show_communication=False, show_log_messages=False - ) + self.interface = OpenMicroStageInterface(show_communication=False, show_log_messages=False) # default connect for the majority of commands self.mock_serial_instance.set_response( @@ -150,9 +145,7 @@ def test_initialization(self): def test_initialization_with_params(self): """Test initialization with custom parameters.""" - interface = OpenMicroStageInterface( - show_communication=False, show_log_messages=False - ) + interface = OpenMicroStageInterface(show_communication=False, show_log_messages=False) self.assertFalse(interface.show_communication) self.assertFalse(interface.show_log_messages) @@ -163,7 +156,7 @@ def test_connect_success(self): DESIRED_FIRMWARE_VERSION, ) self.interface.connect("/dev/ttyACM0") - + self.assertIsNotNone(self.interface.serial) self.assertEqual(self.interface.serial.port, "/dev/ttyACM0") @@ -173,9 +166,9 @@ def test_connect_with_custom_baud_rate(self): SerialInterface.ReplyStatus.OK, DESIRED_FIRMWARE_VERSION, ) - + self.interface.connect("/dev/ttyACM0", baud_rate=115200) - + # Verify the mock was called with correct parameters self.assertIsNotNone(self.interface.serial) self.assertEqual(self.interface.serial.baud_rate, 115200) @@ -186,9 +179,9 @@ def test_connect_incompatible_firmware(self): SerialInterface.ReplyStatus.OK, "v0.9.0", ) - + self.interface.connect("/dev/ttyACM0") - + # Serial should be set to None on incompatible version self.assertIsNone(self.interface.serial) @@ -198,10 +191,10 @@ def test_disconnect(self): SerialInterface.ReplyStatus.OK, DESIRED_FIRMWARE_VERSION, ) - + self.interface.connect("/dev/ttyACM0") self.interface.disconnect() - + self.assertIsNone(self.interface.serial) def test_disconnect_when_not_connected(self): @@ -212,16 +205,11 @@ def test_disconnect_when_not_connected(self): def test_set_and_get_workspace_transform(self): """Test setting and getting workspace transform.""" - transform = np.array([ - [1, 0, 0, 1], - [0, 1, 0, 2], - [0, 0, 1, 3], - [0, 0, 0, 1] - ]) - + transform = np.array([[1, 0, 0, 1], [0, 1, 0, 2], [0, 0, 1, 3], [0, 0, 0, 1]]) + self.interface.set_workspace_transform(transform) result = self.interface.get_workspace_transform() - + self.assertTrue(np.array_equal(result, transform)) def test_read_firmware_version(self): @@ -230,9 +218,9 @@ def test_read_firmware_version(self): SerialInterface.ReplyStatus.OK, "v1.2.3", ) - + major, minor, patch = self.interface.read_firmware_version() - + self.assertEqual(major, 1) self.assertEqual(minor, 2) self.assertEqual(patch, 3) @@ -244,9 +232,9 @@ def test_read_firmware_version_error(self): SerialInterface.ReplyStatus.ERROR, "", ) - + major, minor, patch = self.interface.read_firmware_version() - + self.assertEqual((major, minor, patch), (0, 0, 0)) def test_home_all_axes(self): @@ -255,9 +243,9 @@ def test_home_all_axes(self): SerialInterface.ReplyStatus.OK, "", ) - + result = self.interface.home() - + self.assertEqual(result, SerialInterface.ReplyStatus.OK) self.interface.serial.assert_command_called("G28 A B C D E F\n") @@ -267,9 +255,9 @@ def test_home_specific_axes(self): SerialInterface.ReplyStatus.OK, "", ) - + result = self.interface.home(axis_list=[0, 2]) - + self.assertEqual(result, SerialInterface.ReplyStatus.OK) self.interface.serial.assert_command_called("G28 A C\n") @@ -285,9 +273,9 @@ def test_calibrate_joint_no_save(self): SerialInterface.ReplyStatus.OK, calibration_response, ) - + result, data = self.interface.calibrate_joint(0, save_result=False) - + self.assertEqual(result, SerialInterface.ReplyStatus.OK) self.assertEqual(len(data), 3) self.assertEqual(data[0], [0.5, 1.0, 1.5]) # motor angles @@ -302,9 +290,9 @@ def test_calibrate_joint_with_save(self): SerialInterface.ReplyStatus.OK, calibration_response, ) - + result, data = self.interface.calibrate_joint(1, save_result=True) - + self.assertEqual(result, SerialInterface.ReplyStatus.OK) self.interface.serial.assert_command_called("M56 J1 P S") @@ -314,9 +302,9 @@ def test_read_current_position(self): SerialInterface.ReplyStatus.OK, "X10.5 Y20.3 Z15.8", ) - + x, y, z = self.interface.read_current_position() - + self.assertAlmostEqual(x, 10.5) self.assertAlmostEqual(y, 20.3) self.assertAlmostEqual(z, 15.8) @@ -328,9 +316,9 @@ def test_read_current_position_error(self): SerialInterface.ReplyStatus.ERROR, "", ) - + x, y, z = self.interface.read_current_position() - + self.assertIsNone(x) self.assertIsNone(y) self.assertIsNone(z) @@ -341,7 +329,7 @@ def test_read_current_position_invalid_format(self): SerialInterface.ReplyStatus.OK, "invalid format", ) - + with self.assertRaises(ValueError): self.interface.read_current_position() @@ -351,9 +339,9 @@ def test_move_to_immediate(self): SerialInterface.ReplyStatus.OK, "", ) - + result = self.interface.move_to(5.0, 10.0, 15.0, f=20.0, move_immediately=True) - + self.assertEqual(result, SerialInterface.ReplyStatus.OK) cmd = self.interface.serial.get_last_command() self.assertIn("G0 X5.000000 Y10.000000 Z15.000000 F20.000", cmd) @@ -362,21 +350,16 @@ def test_move_to_immediate(self): def test_move_to_with_workspace_transform(self): """Test move_to applies workspace transform correctly.""" # Set a simple translation transform - transform = np.array([ - [1, 0, 0, 2], - [0, 1, 0, 3], - [0, 0, 1, 4], - [0, 0, 0, 1] - ]) + transform = np.array([[1, 0, 0, 2], [0, 1, 0, 3], [0, 0, 1, 4], [0, 0, 0, 1]]) self.interface.set_workspace_transform(transform) - + self.interface.serial.set_response( SerialInterface.ReplyStatus.OK, "", ) - + result = self.interface.move_to(0, 0, 0, f=10.0) - + # Expected transformed position is (2, 3, 4) cmd = self.interface.serial.get_last_command() self.assertIn("X2.000000", cmd) @@ -385,15 +368,17 @@ def test_move_to_with_workspace_transform(self): def test_move_to_blocking_busy_retry(self): """Test move_to retries on BUSY when blocking is True.""" - self.interface.serial.set_responses([ - (SerialInterface.ReplyStatus.BUSY, ""), - (SerialInterface.ReplyStatus.OK, ""), - ]) - + self.interface.serial.set_responses( + [ + (SerialInterface.ReplyStatus.BUSY, ""), + (SerialInterface.ReplyStatus.OK, ""), + ] + ) + result = self.interface.move_to(5.0, 10.0, 15.0, f=20.0, blocking=True, timeout=0.01) - + self.assertEqual(result, SerialInterface.ReplyStatus.OK) - self.assertEqual(len(self.interface.serial.call_history), 2+self.calls_for_initailization) + self.assertEqual(len(self.interface.serial.call_history), 2 + self.calls_for_initailization) def test_move_to_non_blocking_returns_busy(self): """Test move_to returns BUSY immediately when blocking is False.""" @@ -401,11 +386,11 @@ def test_move_to_non_blocking_returns_busy(self): SerialInterface.ReplyStatus.BUSY, "", ) - + result = self.interface.move_to(5.0, 10.0, 15.0, f=20.0, blocking=False) - + self.assertEqual(result, SerialInterface.ReplyStatus.BUSY) - self.assertEqual(len(self.interface.serial.call_history), 1+self.calls_for_initailization) + self.assertEqual(len(self.interface.serial.call_history), 1 + self.calls_for_initailization) def test_dwell(self): """Test dwell command.""" @@ -413,9 +398,9 @@ def test_dwell(self): SerialInterface.ReplyStatus.OK, "", ) - + result = self.interface.dwell(time_s=2.5, blocking=True) - + self.assertEqual(result, SerialInterface.ReplyStatus.OK) cmd = self.interface.serial.get_last_command() self.assertIn("G4 S2.500000", cmd) @@ -426,11 +411,9 @@ def test_set_max_acceleration(self): SerialInterface.ReplyStatus.OK, "", ) - - result = self.interface.set_max_acceleration( - linear_accel=100.0, angular_accel=50.0 - ) - + + result = self.interface.set_max_acceleration(linear_accel=100.0, angular_accel=50.0) + self.assertEqual(result, SerialInterface.ReplyStatus.OK) cmd = self.interface.serial.get_last_command() self.assertIn("M204 L100.000000 A50.000000", cmd) @@ -441,11 +424,9 @@ def test_set_max_acceleration_minimum_values(self): SerialInterface.ReplyStatus.OK, "", ) - - result = self.interface.set_max_acceleration( - linear_accel=0.001, angular_accel=0.001 - ) - + + self.interface.set_max_acceleration(linear_accel=0.001, angular_accel=0.001) + cmd = self.interface.serial.get_last_command() # Should be clamped to 0.01 self.assertIn("M204 L0.010000 A0.010000", cmd) @@ -456,24 +437,26 @@ def test_wait_for_stop_ready(self): SerialInterface.ReplyStatus.OK, "1", ) - + result = self.interface.wait_for_stop() - + self.assertEqual(result, SerialInterface.ReplyStatus.OK) self.interface.serial.assert_command_called("M53\n") def test_wait_for_stop_polls_until_ready(self): """Test wait_for_stop polls until device is ready.""" - self.interface.serial.set_responses([ - (SerialInterface.ReplyStatus.OK, "0"), - (SerialInterface.ReplyStatus.OK, "0"), - (SerialInterface.ReplyStatus.OK, "1"), - ]) - + self.interface.serial.set_responses( + [ + (SerialInterface.ReplyStatus.OK, "0"), + (SerialInterface.ReplyStatus.OK, "0"), + (SerialInterface.ReplyStatus.OK, "1"), + ] + ) + result = self.interface.wait_for_stop() - + self.assertEqual(result, SerialInterface.ReplyStatus.OK) - self.assertEqual(len(self.interface.serial.call_history), 3+self.calls_for_initailization) + self.assertEqual(len(self.interface.serial.call_history), 3 + self.calls_for_initailization) def test_wait_for_stop_error(self): """Test wait_for_stop returns error status.""" @@ -481,9 +464,9 @@ def test_wait_for_stop_error(self): SerialInterface.ReplyStatus.ERROR, "", ) - + result = self.interface.wait_for_stop() - + self.assertEqual(result, SerialInterface.ReplyStatus.ERROR) def test_enable_motors(self): @@ -492,9 +475,9 @@ def test_enable_motors(self): SerialInterface.ReplyStatus.OK, "", ) - + result = self.interface.enable_motors(enable=True) - + self.assertEqual(result, SerialInterface.ReplyStatus.OK) self.interface.serial.assert_command_called_with_args("M17", timeout=5) @@ -504,9 +487,9 @@ def test_disable_motors(self): SerialInterface.ReplyStatus.OK, "", ) - + result = self.interface.enable_motors(enable=False) - + self.assertEqual(result, SerialInterface.ReplyStatus.OK) self.interface.serial.assert_command_called_with_args("M18", timeout=5) @@ -516,9 +499,9 @@ def test_set_pose(self): SerialInterface.ReplyStatus.OK, "", ) - + result = self.interface.set_pose(x=5.0, y=10.0, z=15.0) - + self.assertEqual(result, SerialInterface.ReplyStatus.OK) cmd = self.interface.serial.get_last_command() self.assertIn("G24 X5.000000 Y10.000000 Z15.000000", cmd) @@ -529,9 +512,9 @@ def test_send_custom_command(self): SerialInterface.ReplyStatus.OK, "response data", ) - + result, response = self.interface.send_command("M57", timeout_s=3.0) - + self.assertEqual(result, SerialInterface.ReplyStatus.OK) self.assertEqual(response, "response data") @@ -541,9 +524,9 @@ def test_read_device_state_info(self): SerialInterface.ReplyStatus.OK, "state info", ) - + result = self.interface.read_device_state_info() - + self.assertEqual(result, SerialInterface.ReplyStatus.OK) self.interface.serial.assert_command_called("M57") @@ -553,9 +536,9 @@ def test_set_servo_parameter_defaults(self): SerialInterface.ReplyStatus.OK, "", ) - + result = self.interface.set_servo_parameter() - + self.assertEqual(result, SerialInterface.ReplyStatus.OK) cmd = self.interface.serial.get_last_command() self.assertIn("M55", cmd) @@ -571,11 +554,11 @@ def test_set_servo_parameter_custom(self): SerialInterface.ReplyStatus.OK, "", ) - + result = self.interface.set_servo_parameter( pos_kp=200, pos_ki=60000, vel_kp=0.3, vel_ki=120, vel_filter_tc=0.003 ) - + self.assertEqual(result, SerialInterface.ReplyStatus.OK) cmd = self.interface.serial.get_last_command() self.assertIn("A200.000000", cmd) @@ -590,9 +573,9 @@ def test_read_encoder_angles(self): SerialInterface.ReplyStatus.OK, "", ) - + result = self.interface.read_encoder_angles() - + self.assertEqual(result, []) self.interface.serial.assert_command_called("M51") @@ -600,7 +583,7 @@ def test_parse_table_data(self): """Test parsing table data.""" data_string = "1.0,2.0,3.0\n4.0,5.0,6.0\n7.0,8.0,9.0" result = OpenMicroStageInterface._parse_table_data(data_string, 3) - + self.assertEqual(len(result), 3) self.assertEqual(result[0], [1.0, 4.0, 7.0]) self.assertEqual(result[1], [2.0, 5.0, 8.0]) @@ -610,7 +593,7 @@ def test_parse_table_data_with_malformed_lines(self): """Test parsing table data skips malformed lines.""" data_string = "1.0,2.0,3.0\ninvalid\n4.0,5.0,6.0" result = OpenMicroStageInterface._parse_table_data(data_string, 3) - + self.assertEqual(len(result), 3) self.assertEqual(result[0], [1.0, 4.0]) self.assertEqual(result[1], [2.0, 5.0]) @@ -620,7 +603,7 @@ def test_parse_table_data_single_row(self): """Test parsing single row of data.""" data_string = "10.5,20.3,15.8" result = OpenMicroStageInterface._parse_table_data(data_string, 3) - + self.assertEqual(result[0], [10.5]) self.assertEqual(result[1], [20.3]) self.assertEqual(result[2], [15.8]) @@ -628,43 +611,48 @@ def test_parse_table_data_single_row(self): def test_workflow_connect_home_move_stop(self): """Test end-to-end workflow: connect, home, move, wait for stop.""" # Set responses for each command in sequence - self.mock_serial_instance.set_responses([ - (SerialInterface.ReplyStatus.OK, DESIRED_FIRMWARE_VERSION), # firmware version - (SerialInterface.ReplyStatus.OK, ""), # home - (SerialInterface.ReplyStatus.OK, ""), # move_to - (SerialInterface.ReplyStatus.OK, "1"), # wait_for_stop - ]) - + self.mock_serial_instance.set_responses( + [ + (SerialInterface.ReplyStatus.OK, DESIRED_FIRMWARE_VERSION), # firmware version + (SerialInterface.ReplyStatus.OK, ""), # home + (SerialInterface.ReplyStatus.OK, ""), # move_to + (SerialInterface.ReplyStatus.OK, "1"), # wait_for_stop + ] + ) + self.interface.connect("/dev/ttyACM0") self.assertIsNotNone(self.interface.serial) - + home_result = self.interface.home() self.assertEqual(home_result, SerialInterface.ReplyStatus.OK) - + move_result = self.interface.move_to(5.0, 10.0, 15.0, f=20.0) self.assertEqual(move_result, SerialInterface.ReplyStatus.OK) - + stop_result = self.interface.wait_for_stop() self.assertEqual(stop_result, SerialInterface.ReplyStatus.OK) def test_workflow_calibrate_and_move(self): """Test end-to-end workflow: calibrate joint and then move.""" calibration_data = "0.5,1.0,100\n1.0,2.0,200\n" - - self.mock_serial_instance.set_responses([ - (SerialInterface.ReplyStatus.OK, calibration_data), # calibrate - (SerialInterface.ReplyStatus.OK, "X5.0 Y10.0 Z15.0"), # read position - ]) - + + self.mock_serial_instance.set_responses( + [ + (SerialInterface.ReplyStatus.OK, calibration_data), # calibrate + (SerialInterface.ReplyStatus.OK, "X5.0 Y10.0 Z15.0"), # read position + ] + ) + result, data = self.interface.calibrate_joint(0, save_result=True) self.assertEqual(result, SerialInterface.ReplyStatus.OK) self.assertEqual(len(data), 3) - + x, y, z = self.interface.read_current_position() self.assertAlmostEqual(x, 5.0) self.assertAlmostEqual(y, 10.0) self.assertAlmostEqual(z, 15.0) +# Manually run unittest if __name__ == "__main__": unittest.main() From d289e62d6bca7164c8d9d1059e94f5948b02c5c6 Mon Sep 17 00:00:00 2001 From: Michael Honaker Date: Mon, 9 Feb 2026 15:59:48 -0500 Subject: [PATCH 10/14] Updates with install instructions --- README.md | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 467db66..1554f02 100644 --- a/README.md +++ b/README.md @@ -44,7 +44,12 @@ It has simple controlls to move the device around, while also displaying a live The lightweight Python API handles all serial communication and provides convenient command execution and debug message printing. The interface includes functions to home, move, and calibrate the device, as well as to query device information. -Simply copy the [open_micro_stage_api.py](software/PythonAPI/open_micro_stage_api.py) file into your project (also install the dependencies in requirements.txt), and you’re ready to get started. + +This library has not been published to pypi yet but it can be installed by running the following pip command. This can also be added to your projects requirements.txt: + +```bash +pip3 install "git+https://github.com/0x23/MicroManipulatorStepper/#subdirectory=software/PythonAPI" +``` ## Usage Example ```python From 833b0804ab58d5696ebbce0e634a98cf06334f40 Mon Sep 17 00:00:00 2001 From: Michael Honaker Date: Mon, 9 Feb 2026 16:02:07 -0500 Subject: [PATCH 11/14] Remove unused utils --- software/PythonAPI/open_micro_stage/utils.py | 36 -------------------- 1 file changed, 36 deletions(-) delete mode 100644 software/PythonAPI/open_micro_stage/utils.py diff --git a/software/PythonAPI/open_micro_stage/utils.py b/software/PythonAPI/open_micro_stage/utils.py deleted file mode 100644 index 57bcabe..0000000 --- a/software/PythonAPI/open_micro_stage/utils.py +++ /dev/null @@ -1,36 +0,0 @@ -type LocalArray = list[float] -type LocalMatrix = list[LocalArray] - - -def generate_eye(size: int) -> LocalMatrix: - """Generate a 2D list representing an eye pattern.""" - return [[1 if i == j else 0 for j in range(size)] for i in range(size)] - - -def matrix_multiply(A: LocalMatrix, B: LocalMatrix) -> LocalMatrix: - """Multiply two matrices A and B.""" - if len(A[0]) != len(B): - raise ValueError("Number of columns in A must equal number of rows in B.") - - result = [[0 for _ in range(len(B[0]))] for _ in range(len(A))] - - for i in range(len(A)): - for j in range(len(B[0])): - for k in range(len(B)): - result[i][j] += A[i][k] * B[k][j] - - return result - - -def matrix_dot_product(A: LocalMatrix, B: LocalArray) -> LocalArray: - """Calculate the dot product of two matrices A and B.""" - if len(A[0]) != len(B): - raise ValueError("Number of columns in A must equal number of elements in B.") - - result = [0 for _ in range(len(A))] - - for i in range(len(A)): - for j in range(len(B)): - result[i] += A[i][j] * B[j] - - return result From ff7b0b3c403af5b8947d8341e35716b7122488d8 Mon Sep 17 00:00:00 2001 From: Michael Honaker Date: Mon, 9 Feb 2026 16:03:12 -0500 Subject: [PATCH 12/14] Update with optional dependencies --- README.md | 2 ++ software/PythonAPI/pyproject.toml | 3 --- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 1554f02..4482b88 100644 --- a/README.md +++ b/README.md @@ -51,6 +51,8 @@ This library has not been published to pypi yet but it can be installed by runni pip3 install "git+https://github.com/0x23/MicroManipulatorStepper/#subdirectory=software/PythonAPI" ``` +If you would like to use the calibration plotter then you will need the additional `plotter` optional dependency. + ## Usage Example ```python from open_micro_stage_api import OpenMicroStageInterface diff --git a/software/PythonAPI/pyproject.toml b/software/PythonAPI/pyproject.toml index 8eee27a..daf3600 100644 --- a/software/PythonAPI/pyproject.toml +++ b/software/PythonAPI/pyproject.toml @@ -28,9 +28,6 @@ build-backend = "setuptools.build_meta" plotter = [ "matplotlib", ] -pretty = [ - "colorama", -] dev = [ "pytest>=7.0", "pytest-cov", From ad6c64ca1ab6f62028122071be523d1ca0e7e8f0 Mon Sep 17 00:00:00 2001 From: Michael Honaker Date: Mon, 9 Feb 2026 16:08:03 -0500 Subject: [PATCH 13/14] Remove unused README.md --- software/PythonAPI/README.md | 79 ------------------------------------ 1 file changed, 79 deletions(-) delete mode 100644 software/PythonAPI/README.md diff --git a/software/PythonAPI/README.md b/software/PythonAPI/README.md deleted file mode 100644 index 475fc73..0000000 --- a/software/PythonAPI/README.md +++ /dev/null @@ -1,79 +0,0 @@ -# Open Micro Stage Python API - -A Python API for controlling the Open Micro Stage micro-manipulator. - -## Installation - -### Development Installation - -```bash -pip install -e . -``` - -### Production Installation - -```bash -pip install open_micro_stage -``` - -## Usage - -```python -from open_micro_stage import OpenMicroStageInterface - -# Create interface and connect -oms = OpenMicroStageInterface(show_communication=True, show_log_messages=True) -oms.connect('/dev/ttyACM0') - -# Home device -oms.home() - -# Move to position -oms.move_to(0, 0, 0, f=10) -oms.wait_for_stop() - -# Read device state -oms.read_device_state_info() -``` - -See `examples/` directory for more usage examples. - -## Requirements - -- Python >= 3.8 -- numpy -- pyserial -- colorama - -## Development - -### Install Development Dependencies - -```bash -pip install -e ".[dev]" -``` - -### Run Tests - -```bash -pytest -``` - -### Code Quality - -The project uses [Ruff](https://github.com/astral-sh/ruff) for: -- Code formatting -- Import sorting -- Linting (pycodestyle, Pyflakes, flake8-bugbear, flake8-comprehensions) - -#### Format and lint code - -```bash -ruff format . -ruff check --fix . -``` - -## License - -See LICENSE file for details. - From c658df2c9a06e33e83a834ebf4969f5ced67a2dc Mon Sep 17 00:00:00 2001 From: Michael Honaker Date: Mon, 9 Feb 2026 16:29:19 -0500 Subject: [PATCH 14/14] Rename lib for compat --- software/PythonAPI/examples/usage_example.py | 2 +- .../{open_micro_stage => open_micro_stage_api}/__init__.py | 0 .../{open_micro_stage => open_micro_stage_api}/__main__.py | 0 .../{open_micro_stage => open_micro_stage_api}/api.py | 0 .../calibration_plotter.py | 0 software/PythonAPI/pyproject.toml | 4 ++-- software/PythonAPI/tests/test_open_micro_stage.py | 2 +- 7 files changed, 4 insertions(+), 4 deletions(-) rename software/PythonAPI/{open_micro_stage => open_micro_stage_api}/__init__.py (100%) rename software/PythonAPI/{open_micro_stage => open_micro_stage_api}/__main__.py (100%) rename software/PythonAPI/{open_micro_stage => open_micro_stage_api}/api.py (100%) rename software/PythonAPI/{open_micro_stage => open_micro_stage_api}/calibration_plotter.py (100%) diff --git a/software/PythonAPI/examples/usage_example.py b/software/PythonAPI/examples/usage_example.py index 0e8bce6..ce3e557 100644 --- a/software/PythonAPI/examples/usage_example.py +++ b/software/PythonAPI/examples/usage_example.py @@ -1,4 +1,4 @@ -from open_micro_stage import OpenMicroStageInterface +from open_micro_stage_api import OpenMicroStageInterface # create interface and connect oms = OpenMicroStageInterface(show_communication=True, show_log_messages=True) diff --git a/software/PythonAPI/open_micro_stage/__init__.py b/software/PythonAPI/open_micro_stage_api/__init__.py similarity index 100% rename from software/PythonAPI/open_micro_stage/__init__.py rename to software/PythonAPI/open_micro_stage_api/__init__.py diff --git a/software/PythonAPI/open_micro_stage/__main__.py b/software/PythonAPI/open_micro_stage_api/__main__.py similarity index 100% rename from software/PythonAPI/open_micro_stage/__main__.py rename to software/PythonAPI/open_micro_stage_api/__main__.py diff --git a/software/PythonAPI/open_micro_stage/api.py b/software/PythonAPI/open_micro_stage_api/api.py similarity index 100% rename from software/PythonAPI/open_micro_stage/api.py rename to software/PythonAPI/open_micro_stage_api/api.py diff --git a/software/PythonAPI/open_micro_stage/calibration_plotter.py b/software/PythonAPI/open_micro_stage_api/calibration_plotter.py similarity index 100% rename from software/PythonAPI/open_micro_stage/calibration_plotter.py rename to software/PythonAPI/open_micro_stage_api/calibration_plotter.py diff --git a/software/PythonAPI/pyproject.toml b/software/PythonAPI/pyproject.toml index daf3600..b6e4c0b 100644 --- a/software/PythonAPI/pyproject.toml +++ b/software/PythonAPI/pyproject.toml @@ -1,5 +1,5 @@ [project] -name = "open_micro_stage" +name = "open_micro_stage_api" version = "0.1.0" description = "Python API for controlling the Open Micro Stage manipulator" readme = "README.md" @@ -41,7 +41,7 @@ Documentation = "https://github.com/HonakerM/MicroManipulatorStepper" Issues = "https://github.com/HonakerM/MicroManipulatorStepper/issues" [tool.setuptools] -packages = ["open_micro_stage"] +packages = ["open_micro_stage_api"] [tool.ruff] line-length = 120 diff --git a/software/PythonAPI/tests/test_open_micro_stage.py b/software/PythonAPI/tests/test_open_micro_stage.py index e502275..2ae0dac 100644 --- a/software/PythonAPI/tests/test_open_micro_stage.py +++ b/software/PythonAPI/tests/test_open_micro_stage.py @@ -5,7 +5,7 @@ import numpy as np -from open_micro_stage.api import OpenMicroStageInterface, SerialInterface +from open_micro_stage_api.api import OpenMicroStageInterface, SerialInterface DESIRED_FIRMWARE_VERSION = "v1.0.1"