From e641dcf91028096583d4a9b0edb4ba52c5e96b25 Mon Sep 17 00:00:00 2001 From: facu Date: Fri, 8 May 2026 17:02:17 -0300 Subject: [PATCH] Bring the latest stage API changes. Updated the test scripts to use it without hardcoded ports. Small Readme for them. --- .gitignore | 4 + software/PythonAPI/README.md | 66 ++++++++++++ software/PythonAPI/calibration_plotter.py | 101 +++++++++++++++--- software/PythonAPI/open_micro_stage_api.py | 113 ++++++++++++++++++--- software/PythonAPI/requirements.txt | 1 + software/PythonAPI/usage_example.py | 101 +++++++++++++++--- 6 files changed, 346 insertions(+), 40 deletions(-) create mode 100644 .gitignore create mode 100644 software/PythonAPI/README.md diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..02c16a3 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +__pycache__/ +*.py[cod] +.DS_Store +._* diff --git a/software/PythonAPI/README.md b/software/PythonAPI/README.md new file mode 100644 index 0000000..ad855b0 --- /dev/null +++ b/software/PythonAPI/README.md @@ -0,0 +1,66 @@ +# Python API + +This folder contains a lightweight Python interface for the Open Micro-Manipulator serial protocol, plus two small example scripts: + +- `usage_example.py`: homes the device, performs a simple move, and prints device state information. +- `calibration_plotter.py`: runs joint calibration for the first three actuators and plots the returned data. + +## Requirements + +Install the Python dependencies with: + +```bash +python -m venv env +source env/bin/activate +pip install -r requirements.txt +``` + +The API itself is implemented in `open_micro_stage_api.py`. + +## Serial Port Selection + +Both scripts support: + +- `--list-ports`: list detected serial devices and exit +- `--port `: explicitly select a serial port + +If `--port` is not provided, the scripts try to choose a port automatically: + +1. If exactly one detected device contains `Pico` in its name, that port is used. +2. Otherwise, if there is exactly one detected serial device, that port is used. +3. Otherwise, the script lists the available ports and exits. + +## Running The Example Script + +From this folder: + +```bash +python usage_example.py --list-ports +python usage_example.py --port /dev/ttyACM0 +``` + +On Windows, a typical command looks like: + +```bash +python usage_example.py --port COM3 +``` + +## Running The Calibration Plotter + +From this folder: + +```bash +python calibration_plotter.py --list-ports +python calibration_plotter.py --port /dev/ttyACM0 +``` + +The calibration script opens a matplotlib window with the measured calibration curves. + +## Running From The Repository Root + +If you prefer to run the scripts from the repository root, use: + +```bash +python software/PythonAPI/usage_example.py --port /dev/ttyACM0 +python software/PythonAPI/calibration_plotter.py --port /dev/ttyACM0 +``` diff --git a/software/PythonAPI/calibration_plotter.py b/software/PythonAPI/calibration_plotter.py index c0e3f26..a6851de 100644 --- a/software/PythonAPI/calibration_plotter.py +++ b/software/PythonAPI/calibration_plotter.py @@ -1,7 +1,12 @@ -from open_micro_stage_api import OpenMicroStageInterface +import argparse + import matplotlib.pyplot as plt + +from open_micro_stage_api 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: @@ -21,20 +26,92 @@ def plot_calibration_data(ax_encoder_counts, ax_field_angel, label, data): ax_field_angel.legend() ax_field_angel.grid(True) + +def list_available_ports(): + devices = OpenMicroStageInterface.enumerate_devices() + if not devices: + print('No serial devices detected.') + return devices + + print('Available serial devices:') + for device in devices: + print(f" {device['label']}") + + return devices + + +def resolve_port(port): + if port: + return port + + devices = OpenMicroStageInterface.enumerate_devices() + pico_devices = [device for device in devices if 'pico' in device['label'].lower()] + + if len(pico_devices) == 1: + print(f"Using detected Pico serial device: {pico_devices[0]['label']}") + return pico_devices[0]['port'] + + if len(pico_devices) > 1: + print('Multiple Pico serial devices detected. Pass --port to choose one:') + for device in pico_devices: + print(f" {device['label']}") + raise SystemExit(1) + + if len(devices) == 1: + print(f"Using detected serial device: {devices[0]['label']}") + return devices[0]['port'] + + if devices: + print('Multiple serial devices detected. Pass --port to choose one:') + for device in devices: + print(f" {device['label']}") + else: + print('No serial devices detected.') + + raise SystemExit(1) + + +def parse_args(): + parser = argparse.ArgumentParser( + description='Run joint calibration and plot the measured data.', + ) + parser.add_argument( + '--port', + help='Serial port to use (for example /dev/ttyACM0 or COM3).', + ) + parser.add_argument( + '--list-ports', + action='store_true', + help='List detected serial devices and exit.', + ) + return parser.parse_args() + + def main(): - # create interface and connect + args = parse_args() + if args.list_ports: + list_available_ports() + return + + port = resolve_port(args.port) oms = OpenMicroStageInterface(show_communication=True, show_log_messages=True) - oms.connect('/dev/ttyACM0') + if not oms.connect(port): + raise SystemExit(f'Could not connect to {port}.') - # Create subplots - fig, ax = plt.subplots(1, 1, figsize=(10, 7), sharex='all') + try: + # Create subplots + 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) + for i in range(3): + res, data = oms.calibrate_joint(i, save_result=False) + plot_calibration_data(ax, None, f'Actuator {i}', data) - # Adjust layout and show - plt.tight_layout() - plt.show() + # Adjust layout and show + plt.tight_layout() + plt.show() + finally: + oms.disconnect() -main() \ No newline at end of file + +if __name__ == '__main__': + main() diff --git a/software/PythonAPI/open_micro_stage_api.py b/software/PythonAPI/open_micro_stage_api.py index 3dc8835..6e9c1b1 100644 --- a/software/PythonAPI/open_micro_stage_api.py +++ b/software/PythonAPI/open_micro_stage_api.py @@ -1,3 +1,10 @@ +# -------------------------------------------------------------------------------------- +# Project: OpenMicroManipulator +# License: MIT (see LICENSE file for full description) +# All text in here must be included in any redistribution. +# Author: M. S. (diffraction limited) +# -------------------------------------------------------------------------------------- + import threading import time import re @@ -6,6 +13,7 @@ from enum import Enum import serial import numpy as np from colorama import Fore, Style, init +from serial.tools import list_ports # --- SerialInterface -------------------------------------------------------------------------------------------------- @@ -59,18 +67,23 @@ class SerialInterface: self._response_string = "" self._response_status = None self._response_error_msg = None + self._stop_event = threading.Event() + self._reader_thread = None - self.connect(self.reconnect_timeout) - - # Start reader thread - self._reader_thread = threading.Thread(target=self._reader_loop, daemon=True) - self._reader_thread.start() + if self.connect(self.reconnect_timeout): + self._reader_thread = threading.Thread(target=self._reader_loop, daemon=True) + self._reader_thread.start() + while not self._reader_thread.is_alive(): + time.sleep(0.001) def connect(self, timeout): """ Try to open the serial port. Retry until timeout expires. """ + if self._stop_event.is_set(): + return False + deadline = time.time() + timeout print(Fore.MAGENTA, end='') print(f"[SerialInterface] Connecting to port '{self.port}'...", end='') @@ -95,7 +108,7 @@ class SerialInterface: Asynchronous reader loop, collecting serial data into a buffer """ buffer = "" - while True: + while not self._stop_event.is_set(): try: if self.serial is not None and self.serial.in_waiting: char = self.serial.read(1).decode('ascii', errors='ignore') @@ -108,6 +121,9 @@ class SerialInterface: else: time.sleep(0.001) except (serial.SerialException, OSError) as e: + if self._stop_event.is_set(): + break + print(Fore.MAGENTA+f"[SerialInterface] Lost connection: {e}"+Style.RESET_ALL) try: if self.serial is not None and self.serial.is_open: @@ -195,8 +211,13 @@ class SerialInterface: def close(self): """Closes the serial port.""" + self._stop_event.set() if self.serial and self.serial.is_open: self.serial.close() + self.serial = None + + if self._reader_thread and self._reader_thread.is_alive(): + self._reader_thread.join(timeout=1.0) # --- OpenMicroStageInterface ------------------------------------------------------------------------------------------ @@ -212,19 +233,45 @@ class OpenMicroStageInterface: def __init__(self, show_communication=True, show_log_messages=True): self.serial = None self.workspace_transform = np.eye(4) + self.workspace_transform_inv = np.linalg.inv(self.workspace_transform) self.show_communication = show_communication self.show_log_messages = show_log_messages self.disable_message_callbacks = False + @staticmethod + def enumerate_devices(): + devices = [] + + for port in sorted(list_ports.comports(), key=lambda item: item.device): + description = (port.description or "").strip() + label = port.device + if description and description.lower() != "n/a" and description != port.device: + label = f"{port.device} - {description}" + + devices.append({ + "id": port.device, + "label": label, + "port": port.device, + }) + + return devices + 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() + + serial_interface = 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 serial_interface.serial is None: + serial_interface.close() + return False + + self.serial = serial_interface self.disable_message_callbacks = True fw_version = self.read_firmware_version() @@ -233,15 +280,23 @@ class OpenMicroStageInterface: 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) + self.serial.close() self.serial = None + print('') + self.disable_message_callbacks = False + return False print('') self.disable_message_callbacks = False + return True def disconnect(self): if self.serial is not None: self.serial.close() self.serial = None + def is_connected(self): + return self.serial is not None + def log_msg_callback(self, log_level, msg): if not self.show_log_messages or self.disable_message_callbacks: return @@ -273,6 +328,7 @@ class OpenMicroStageInterface: def set_workspace_transform(self, transform): self.workspace_transform = transform + self.workspace_transform_inv = np.linalg.inv(self.workspace_transform) def get_workspace_transform(self): return self.workspace_transform @@ -282,7 +338,11 @@ class OpenMicroStageInterface: 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()) + match = re.match(r'v(\d+)\.(\d+)\.(\d+)', response) + if match is None: + return 0, 0, 0 + + major, minor, patch = map(int, match.groups()) return major,minor,patch def home(self, axis_list=None): @@ -370,11 +430,17 @@ class OpenMicroStageInterface: res, msg = self.serial.send_command("M53\n") if res != SerialInterface.ReplyStatus.OK: return res elif msg.strip() == "1": + self.disable_message_callbacks = disable_message_callbacks_prev return SerialInterface.ReplyStatus.OK + time.sleep(polling_interval_ms*0.001) self.disable_message_callbacks = disable_message_callbacks_prev - def read_current_position(self): + def read_current_position(self, apply_inv_workspace_transform): + """ + Reads the current position of the dives EXCLUDING the workspace transform. + If you want to use the result with a + """ ok, response = self.serial.send_command("M50") if ok != SerialInterface.ReplyStatus.OK or len(response) == 0: return None, None, None @@ -388,6 +454,10 @@ class OpenMicroStageInterface: raise ValueError(f"Invalid format: {response}") x, y, z = match.groups() + if apply_inv_workspace_transform: + transformed = self.workspace_transform_inv @ np.array([float(x), float(y), float(z), 1.0]) + x, y, z = transformed[:3] / transformed[3] + return float(x), float(y), float(z) def read_encoder_angles(self): @@ -398,19 +468,19 @@ class OpenMicroStageInterface: def read_device_state_info(self): res, msg = self.serial.send_command("M57") - return res + return res, msg def set_servo_parameter(self, pos_kp=150, pos_ki=50000, vel_kp=0.2, vel_ki=100, vel_filter_tc=0.0025): cmd = f"M55 A{pos_kp:.6f} B{pos_ki:.6f} C{vel_kp:.6f} D{vel_ki:.6f} F{vel_filter_tc:.6f}" res, msg = self.serial.send_command(cmd) return res - def enable_motors(self, enable): + def enable_motors(self, enable: bool): cmd = f"M17" if enable else "M18" res, msg = self.serial.send_command(cmd, timeout=5) return res - def set_pose(self, x, y, z): + def set_pose(self, x: float, y: float, z: float): # Convert to homogeneous vector transformed = self.workspace_transform @ np.array([x, y, z, 1.0]) x_t, y_t, z_t = transformed[:3] / transformed[3] @@ -419,6 +489,17 @@ class OpenMicroStageInterface: res, msg = self.serial.send_command(cmd) return res + def set_tool_output(self, tool_idx: int, output_value: float, immediate: bool = True): + # sets the output value for the specified tool + cmd = f"M3 T{tool_idx} S{output_value}" + res, msg = self.serial.send_command(cmd) + + # send dwell command to update tool value immediately + if immediate: + self.serial.send_command("G4 S0.001") + + return res + def send_command(self, cmd: str, timeout_s: float=5): res, msg = self.serial.send_command(cmd, timeout_s) return res, msg diff --git a/software/PythonAPI/requirements.txt b/software/PythonAPI/requirements.txt index 6e4d37a..bf33c24 100644 --- a/software/PythonAPI/requirements.txt +++ b/software/PythonAPI/requirements.txt @@ -1,3 +1,4 @@ numpy pyserial colorama +matplotlib diff --git a/software/PythonAPI/usage_example.py b/software/PythonAPI/usage_example.py index 212a305..1484b38 100644 --- a/software/PythonAPI/usage_example.py +++ b/software/PythonAPI/usage_example.py @@ -1,18 +1,95 @@ +import argparse + from open_micro_stage_api import OpenMicroStageInterface -# create interface and connect -oms = OpenMicroStageInterface(show_communication=True, show_log_messages=True) -oms.connect('/dev/ttyACM0') -# run this once to calibrate joints -# for i in range(3): oms.calibrate_joint(i, save_result=True) +def list_available_ports(): + devices = OpenMicroStageInterface.enumerate_devices() + if not devices: + print('No serial devices detected.') + return devices -# home device -oms.home() + print('Available serial devices:') + for device in devices: + print(f" {device['label']}") -# move and wait -oms.move_to(0, 0, 0, f=10) -oms.wait_for_stop() + return devices -# print some info -oms.read_device_state_info() \ No newline at end of file + +def resolve_port(port): + if port: + return port + + devices = OpenMicroStageInterface.enumerate_devices() + pico_devices = [device for device in devices if 'pico' in device['label'].lower()] + + if len(pico_devices) == 1: + print(f"Using detected Pico serial device: {pico_devices[0]['label']}") + return pico_devices[0]['port'] + + if len(pico_devices) > 1: + print('Multiple Pico serial devices detected. Pass --port to choose one:') + for device in pico_devices: + print(f" {device['label']}") + raise SystemExit(1) + + if len(devices) == 1: + print(f"Using detected serial device: {devices[0]['label']}") + return devices[0]['port'] + + if devices: + print('Multiple serial devices detected. Pass --port to choose one:') + for device in devices: + print(f" {device['label']}") + else: + print('No serial devices detected.') + + raise SystemExit(1) + + +def parse_args(): + parser = argparse.ArgumentParser( + description='Basic OpenMicroManipulator serial API example.', + ) + parser.add_argument( + '--port', + help='Serial port to use (for example /dev/ttyACM0 or COM3).', + ) + parser.add_argument( + '--list-ports', + action='store_true', + help='List detected serial devices and exit.', + ) + return parser.parse_args() + + +def main(): + args = parse_args() + if args.list_ports: + list_available_ports() + return + + port = resolve_port(args.port) + oms = OpenMicroStageInterface(show_communication=True, show_log_messages=True) + if not oms.connect(port): + raise SystemExit(f'Could not connect to {port}.') + + try: + # run this once to calibrate joints + # for i in range(3): oms.calibrate_joint(i, save_result=True) + + # home device + oms.home() + + # move and wait + oms.move_to(0, 0, 0, f=10) + oms.wait_for_stop() + + # print some info + oms.read_device_state_info() + finally: + oms.disconnect() + + +if __name__ == '__main__': + main()