From 62c63ef6fff06dc3a0612b13f7dd8fe019f3e399 Mon Sep 17 00:00:00 2001 From: You Yan Date: Fri, 10 Jul 2026 14:18:37 -0700 Subject: [PATCH 1/9] feat(stage): public MS2000Serial accessors + find_shared_ms2000 CombinedStage.z_stage, LS50Controller.serial, ASIZStage.ms2000_serial, and a module-level find_shared_ms2000(stage) so a same-controller addon (the ASI objective turret) can reuse the Z stage's lock-protected transport without reaching through private attrs. Simulation-aware (None for _SimulatedLS50); ownership stays with the Z stage. Co-Authored-By: Claude Fable 5 --- software/squid/stage/asi.py | 24 ++++++++++++++++++++++++ software/squid/stage/pi.py | 5 +++++ software/tests/squid/test_stage.py | 18 ++++++++++++++++++ 3 files changed, 47 insertions(+) diff --git a/software/squid/stage/asi.py b/software/squid/stage/asi.py index 771a20bfb..e78a726fb 100644 --- a/software/squid/stage/asi.py +++ b/software/squid/stage/asi.py @@ -147,6 +147,11 @@ def __init__(self, axis: str = "Z"): self._range_lo: Optional[float] = None self._range_hi: Optional[float] = None + @property + def serial(self) -> Optional[MS2000Serial]: + """The MS-2000 transport (public: shared with same-controller addons, e.g. the turret).""" + return self._serial + def connect_serial(self, comport: str, baudrate: int = 115200) -> None: self._serial = MS2000Serial.open(comport, baudrate=baudrate) @@ -279,6 +284,11 @@ def __init__( self._invert = invert_z self._home_mm = home_mm # Squid-frame retract target; None = home(z) disabled + @property + def ms2000_serial(self) -> Optional[MS2000Serial]: + # getattr: _SimulatedLS50 has no `serial` property -> None (simulation-aware). + return getattr(self._backend, "serial", None) + def _flip(self, mm: float) -> float: # squid_z = -native_z (and vice versa) when inverted; identity otherwise. return -mm if self._invert else mm @@ -383,6 +393,20 @@ def _no_xy(self, name: str): self._log.warning(f"{name} ignored: ASIZStage is a Z-only stage (pair via CombinedStage).") +def find_shared_ms2000(stage) -> Optional[MS2000Serial]: + """Return the MS2000Serial of an ASI Z stage embedded in ``stage``, else None. + + Accepts an ASIZStage directly or a composite exposing a ``z_stage`` property (duck-typed + to avoid an import cycle with squid.stage.pi). Returns None for simulated backends and + non-ASI stages. Ownership stays with the Z stage: callers must NOT close the returned + transport -- it is released by stage.close(). + """ + for candidate in (stage, getattr(stage, "z_stage", None)): + if isinstance(candidate, ASIZStage): + return candidate.ms2000_serial + return None + + def connect_asi_z_stage( simulated: bool = False, serialnum: Optional[str] = None, diff --git a/software/squid/stage/pi.py b/software/squid/stage/pi.py index c26c1a3cc..861e3a540 100644 --- a/software/squid/stage/pi.py +++ b/software/squid/stage/pi.py @@ -298,6 +298,11 @@ def __init__(self, xy_stage: AbstractStage, z_stage: AbstractStage, stage_config ) self._config = self._config.model_copy(update={"Z_AXIS": fine_z}) + @property + def z_stage(self) -> AbstractStage: + """The wrapped Z-only stage (public: addons sharing its transport walk through here).""" + return self._z + def move_x(self, rel_mm: float, blocking: bool = True): self._xy.move_x(rel_mm, blocking) diff --git a/software/tests/squid/test_stage.py b/software/tests/squid/test_stage.py index 5c88b1f30..2605e073a 100644 --- a/software/tests/squid/test_stage.py +++ b/software/tests/squid/test_stage.py @@ -753,3 +753,21 @@ def test_ls50_initialize_retries_once(): ctrl = _ls50_ctrl(conn) ctrl.initialize() # first W Z gets dead air, retry parses assert conn.written == [b"W Z\r", b"W Z\r"] + + +def test_find_shared_ms2000_walks_combined_stage(): + # The turret (same MS-2000 controller) reuses the Z stage's transport; the accessor + # must return the exact MS2000Serial instance through CombinedStage or a bare ASIZStage. + conn = _FakeSerialConn(replies=[b":A 0\r\n"]) + ctrl = _ls50_ctrl(conn) + z = squid.stage.asi.ASIZStage(ctrl, stage_config=squid.config.get_stage_config()) + assert squid.stage.asi.find_shared_ms2000(z) is ctrl.serial + combined, _, _ = _sim_combined_stage(z_stage=z) + assert squid.stage.asi.find_shared_ms2000(combined) is ctrl.serial + + +def test_find_shared_ms2000_none_for_simulated_and_foreign(): + assert squid.stage.asi.find_shared_ms2000(_sim_asi_stage()) is None # _SimulatedLS50 backend + combined, _, _ = _sim_combined_stage() # PI Z, not ASI + assert squid.stage.asi.find_shared_ms2000(combined) is None + assert squid.stage.asi.find_shared_ms2000(None) is None From 9a7aae12878fd591e4a0191f45cf507965e91f89 Mon Sep 17 00:00:00 2001 From: You Yan Date: Fri, 10 Jul 2026 14:24:16 -0700 Subject: [PATCH 2/9] feat(turret): ASI 6-position objective turret (shared MS-2000 with the LS50 Z) control/asi_objective_turret.py: ASIObjectiveTurret + simulation implementing ObjectiveChangerProtocol. Rotates with 'M T=' (RAW slot index 1..6, not scaled like linear-axis units); reads position via best-effort 'W T' (probed at connect, verified after each move) degrading to last-commanded-slot tracking; NO homing exists -- home() only refreshes the tracked slot; Z retract/restore dance on every switch (template semantics); shared-serial ownership: never closes a transport the Z stage owns, closes its own. Wiring: USE_ASI_OBJECTIVE_TURRET + ASI_OBJECTIVE_TURRET_* flags (SN/port/baud default to the ASI_Z_* values, resolved at construction time), 3-way mutual-exclusion validator, MicroscopeAddons elif branch using find_shared_ms2000(stage), GUI cleanup condition widened. 31 new turret tests + 6 config tests, incl. simulated end-to-end Microscope builds and the raw-slot-framing proof. Co-Authored-By: Claude Fable 5 --- software/control/_def.py | 32 +- software/control/asi_objective_turret.py | 309 ++++++++++++++++++ software/control/gui_hcs.py | 4 +- software/control/microscope.py | 31 +- .../control/test_asi_objective_turret.py | 277 ++++++++++++++++ .../control/test_objective_changer_config.py | 24 ++ 6 files changed, 669 insertions(+), 8 deletions(-) create mode 100644 software/control/asi_objective_turret.py create mode 100644 software/tests/control/test_asi_objective_turret.py diff --git a/software/control/_def.py b/software/control/_def.py index e68c47fd0..fb03b04ee 100644 --- a/software/control/_def.py +++ b/software/control/_def.py @@ -1294,11 +1294,31 @@ def get_wellplate_settings(wellplate_format): # Objective name -> turret slot index (1..4). Override per machine in .ini. OBJECTIVE_TURRET_POSITIONS = {"4x": 1, "10x": 2, "20x": 3, "40x": 4} - -def _validate_objective_changer_flags(use_xeryon: bool, use_turret: bool) -> None: - if use_xeryon and use_turret: +# 6-position ASI objective turret on an MS-2000-family controller ('M T=' with a RAW +# slot index 1..6 -- not scaled like linear-axis units). Typically the SAME controller as the +# ASI LS50 Z stage (USE_ASI_Z_STAGE): when both are enabled the turret reuses the Z stage's +# serial connection and the SN/port/baud below are ignored. They apply only when the turret +# runs WITHOUT the ASI Z stage; empty/0 means "reuse the ASI_Z_* value" -- resolved at +# construction time in microscope.py, NOT here, because the machine-config loader overrides +# these globals after definition (same reasoning as uses_external_z_stage()). +# The turret has NO homing: home() never moves; startup's move_to_objective(DEFAULT_OBJECTIVE) +# establishes a known slot at every boot. +USE_ASI_OBJECTIVE_TURRET = False +# Axis letter of the turret on the controller ('M =' / 'W '). +ASI_OBJECTIVE_TURRET_AXIS_LETTER = "T" +# Objective name -> turret slot (1..6). Keys must match objectives.csv names. Override per machine .ini +# (JSON with double-quoted keys, e.g. asi_objective_turret_positions = {"4x": 1, "20x": 2}). +ASI_OBJECTIVE_TURRET_POSITIONS = {"2x": 1, "4x": 2, "10x": 3, "20x": 4, "40x": 5, "60x": 6} +ASI_OBJECTIVE_TURRET_SN = "" # "" = reuse ASI_Z_STAGE_SN (turret-without-Z-stage only) +ASI_OBJECTIVE_TURRET_SERIAL_PORT = "" # "" = reuse ASI_Z_SERIAL_PORT +ASI_OBJECTIVE_TURRET_BAUDRATE = 0 # 0 = reuse ASI_Z_BAUDRATE + + +def _validate_objective_changer_flags(use_xeryon: bool, use_turret: bool, use_asi_turret: bool = False) -> None: + if int(bool(use_xeryon)) + int(bool(use_turret)) + int(bool(use_asi_turret)) > 1: raise ValueError( - "USE_XERYON and USE_OBJECTIVE_TURRET are mutually exclusive " "(set only one to True in the machine .ini)" + "USE_XERYON, USE_OBJECTIVE_TURRET and USE_ASI_OBJECTIVE_TURRET are mutually exclusive " + "(set at most one to True in the machine .ini)" ) @@ -1407,7 +1427,7 @@ class SlackNotifications: myclass = locals()[classkey] populate_class_from_dict(myclass, pop_items) - _validate_objective_changer_flags(USE_XERYON, USE_OBJECTIVE_TURRET) + _validate_objective_changer_flags(USE_XERYON, USE_OBJECTIVE_TURRET, USE_ASI_OBJECTIVE_TURRET) _validate_external_z_stage_flags(USE_PI_FOCUS_STAGE, USE_ASI_Z_STAGE) with open("cache/config_file_path.txt", "w") as file: @@ -1422,7 +1442,7 @@ class SlackNotifications: sys.exit(1) log.info("load machine-specific configuration") exec(open(config_files[0]).read()) - _validate_objective_changer_flags(USE_XERYON, USE_OBJECTIVE_TURRET) + _validate_objective_changer_flags(USE_XERYON, USE_OBJECTIVE_TURRET, USE_ASI_OBJECTIVE_TURRET) _validate_external_z_stage_flags(USE_PI_FOCUS_STAGE, USE_ASI_Z_STAGE) else: log.error("machine-specific configuration not present, the program will exit") diff --git a/software/control/asi_objective_turret.py b/software/control/asi_objective_turret.py new file mode 100644 index 000000000..81ce125f9 --- /dev/null +++ b/software/control/asi_objective_turret.py @@ -0,0 +1,309 @@ +"""6-position ASI objective turret on an MS-2000-family controller. + +Shares the controller -- and, when USE_ASI_Z_STAGE is enabled, the serial connection -- with +the LS50 Z stage (squid.stage.asi). Command set: ``M T=`` rotates to a RAW slot index +1..6 (NOT scaled by the 0.1 um unit factor used for linear axes); ``W T`` presumably reads +the slot back (UNVERIFIED on hardware -- treated as best-effort, degrading to software +tracking of the last commanded slot); ``/`` is the controller-GLOBAL busy byte, shared with +the Z axis. ``MS2000Serial``'s internal lock keeps turret and Z commands frame-safe when +interleaved. + +The turret has NO homing: ``home()`` never moves; it only refreshes the tracked slot. The +startup flow's ``move_to_objective(DEFAULT_OBJECTIVE)`` establishes a known slot at boot. +""" + +import re +import threading +import time +from contextlib import suppress +from typing import Dict, Optional + +import squid.abc +import squid.logging +from control.objective_turret_controller import _resolve_position # same KeyError message the GUI dialog shows +from squid.stage.asi import MS2000Serial + +_log = squid.logging.get_logger(__name__) + +TURRET_SLOT_COUNT = 6 +DEFAULT_MOVE_TIMEOUT_S = 30.0 +_STATUS_POLL_PERIOD_S = 0.05 + + +def _validate_positions(positions: Optional[dict]) -> Dict[str, int]: + if positions is None: + from control._def import ASI_OBJECTIVE_TURRET_POSITIONS + + positions = ASI_OBJECTIVE_TURRET_POSITIONS + # Fail fast: a bad slot value would otherwise go straight to the hardware as 'M T='. + for name, slot in positions.items(): + if not isinstance(slot, int) or not 1 <= slot <= TURRET_SLOT_COUNT: + raise ValueError( + f"ASI turret position for {name!r} must be an integer slot 1..{TURRET_SLOT_COUNT}, got {slot!r}" + ) + return dict(positions) + + +class ASIObjectiveTurret: + """ObjectiveChangerProtocol implementation for the MS-2000 turret (T) axis. + + Pass ``shared_serial`` (the Z stage's transport, via squid.stage.asi.find_shared_ms2000) + when the LS50 Z stage runs on the same controller -- the turret then never closes it. + Without it, the turret opens (and owns) its own connection by serial number or port. + """ + + def __init__( + self, + shared_serial: Optional[MS2000Serial] = None, + serial_number: Optional[str] = None, + serial_port: Optional[str] = None, + baudrate: int = 115200, + axis: str = "T", + positions: Optional[Dict[str, int]] = None, + stage: Optional[squid.abc.AbstractStage] = None, + ): + self._positions = _validate_positions(positions) + self._axis = axis + self._stage = stage + self._current_objective: Optional[str] = None + self._is_open = False + + if shared_serial is not None: + self._serial = shared_serial + self._owns_serial = False + _log.info("ASI turret reusing the Z stage's MS-2000 connection.") + else: + if serial_port: + port = serial_port + elif serial_number: + import squid.stage.utils + + port = squid.stage.utils.resolve_serial_port_by_sn( + serial_number, + missing_hint="Verify the MS-2000 controller is powered and enumerates as a USB serial device.", + ) + else: + raise RuntimeError( + "Set ASI_OBJECTIVE_TURRET_SN/ASI_OBJECTIVE_TURRET_SERIAL_PORT (or the ASI_Z_* " + "equivalents) to locate the MS-2000 controller." + ) + _log.info(f"ASI turret opening its own MS-2000 connection on {port} at {baudrate} baud.") + self._serial = MS2000Serial.open(port, baudrate=baudrate) + self._owns_serial = True + + try: + # require_reply only when this is our own connection (then it doubles as the comms + # sanity check); on a shared connection the Z stage already proved the link. + self._current_slot = self._probe_slot(require_reply=self._owns_serial) + except Exception: + if self._owns_serial: + self._serial.close() # no leaked handle on a failed bring-up + raise + self._is_open = True + + # --- ObjectiveChangerProtocol --------------------------------------------- + + def home(self, timeout_s: float = DEFAULT_MOVE_TIMEOUT_S) -> None: + """The ASI turret has no homing: refresh the tracked slot; NEVER moves.""" + self._require_open() + self._current_objective = None + slot = self._probe_slot() + if slot is not None: + self._current_slot = slot + _log.info("ASI turret has no homing; home() refreshed the tracked slot without motion.") + + def move_to_objective( + self, objective_name: str, timeout_s: float = DEFAULT_MOVE_TIMEOUT_S, restore_z: bool = True + ) -> None: + self._require_open() + target_slot = _resolve_position(objective_name, self._positions) + if self._current_slot == target_slot: + # Already at the slot (alias name, or the W-T-seeded position). An unknown slot + # (None) never equals an int, so uncertainty always rotates. + self._current_objective = objective_name + return + captured_z = self._retract_z_if_possible() + try: + self._rotate_to_slot(target_slot, timeout_s) + self._current_objective = objective_name + finally: + if restore_z: + self._restore_z_if_captured(captured_z) + + def close(self) -> None: + if not self._is_open: + return + self._is_open = False + if self._owns_serial: + self._serial.close() + # A shared transport belongs to the Z stage and is released by stage.close(). + + def __enter__(self) -> "ASIObjectiveTurret": + return self + + def __exit__(self, *exc) -> None: + self.close() + + # --- properties ------------------------------------------------------------- + + @property + def current_objective(self) -> Optional[str]: + return self._current_objective + + @property + def current_slot(self) -> Optional[int]: + return self._current_slot + + @property + def is_open(self) -> bool: + return self._is_open + + @property + def owns_serial(self) -> bool: + return self._owns_serial + + # --- internals --------------------------------------------------------------- + + def _require_open(self) -> None: + if not self._is_open: + raise RuntimeError("ASI turret is closed") + + def _probe_slot(self, require_reply: bool = False) -> Optional[int]: + """Best-effort 'W T' read. Returns the slot only for a plausible 1..N integer reply. + + check_error=False: an ':N-...' ack still proves comms -- it just means 'W T' is + unsupported on this controller build, and we fall back to tracking the last + commanded slot. + """ + reply = self._serial.command(f"W {self._axis}", check_error=False) + if require_reply and not reply: + # Our own connection: this probe doubles as the comms sanity check (mirrors + # LS50Controller.initialize -- settle, flush, one retry). + time.sleep(0.2) + with suppress(Exception): + self._serial._serial.reset_input_buffer() + reply = self._serial.command(f"W {self._axis}", check_error=False) + if not reply: + raise RuntimeError( + f"MS-2000 did not answer 'W {self._axis}'. Check the baud rate (ASI RS-232 " + f"DIP default is 9600), the port, and that the controller is powered." + ) + _log.info(f"ASI turret 'W {self._axis}' reply: {reply!r}") + match = re.search(r"-?\d+", reply.replace(":A", "", 1)) if reply else None + if match: + slot = int(match.group(0)) + if 1 <= slot <= TURRET_SLOT_COUNT: + return slot + return None + + def _rotate_to_slot(self, slot: int, timeout_s: float) -> None: + wt_was_supported = self._current_slot is not None + # RAW slot index -- turret positions are not scaled like linear-axis 0.1 um units. + self._serial.command(f"M {self._axis}={int(slot)}") + self._wait_idle(timeout_s) + self._current_slot = slot + if wt_was_supported: + # Verification only; tracked state stays authoritative. + with suppress(Exception): + readback = self._probe_slot() + if readback is not None and readback != slot: + _log.warning(f"ASI turret readback disagrees: commanded slot {slot}, 'W' reports {readback}.") + + def _wait_idle(self, timeout_s: float) -> None: + # '/' is controller-global: a concurrently moving Z also reads busy. Acceptable -- + # objective changes and Z scans are not concurrent flows. + deadline = time.monotonic() + timeout_s + while "B" in self._serial.command("/").upper(): + if time.monotonic() > deadline: + raise RuntimeError(f"ASI turret did not reach idle within {timeout_s:.1f}s") + time.sleep(_STATUS_POLL_PERIOD_S) + + def _retract_z_if_possible(self) -> Optional[float]: + from control._def import HOMING_ENABLED_Z, OBJECTIVE_RETRACTED_POS_MM + + if self._stage is None or not HOMING_ENABLED_Z: + return None + z_mm = self._stage.get_pos().z_mm + self._stage.move_z_to(OBJECTIVE_RETRACTED_POS_MM) + return z_mm + + def _restore_z_if_captured(self, captured_z: Optional[float]) -> None: + if captured_z is None or self._stage is None: + return + self._stage.move_z_to(captured_z) + + +class ASIObjectiveTurretSimulation: + """Hardware-free stand-in with the same public surface (SIMULATE_OBJECTIVE_CHANGER).""" + + def __init__( + self, + positions: Optional[Dict[str, int]] = None, + stage: Optional[squid.abc.AbstractStage] = None, + axis: str = "T", + ): + self._positions = _validate_positions(positions) + self._axis = axis + self._stage = stage + self._current_objective: Optional[str] = None + self._current_slot: Optional[int] = None # unknown until commanded, like power-on + self._is_open = True + + def home(self, timeout_s: float = DEFAULT_MOVE_TIMEOUT_S) -> None: + self._require_open() + self._current_objective = None # slot is retained: no homing, nothing moved + + def move_to_objective( + self, objective_name: str, timeout_s: float = DEFAULT_MOVE_TIMEOUT_S, restore_z: bool = True + ) -> None: + self._require_open() + target_slot = _resolve_position(objective_name, self._positions) + if self._current_slot == target_slot: + self._current_objective = objective_name + return + captured_z = self._retract_z_if_possible() + try: + self._current_slot = target_slot + self._current_objective = objective_name + finally: + if restore_z: + self._restore_z_if_captured(captured_z) + + def close(self) -> None: + self._is_open = False + + def __enter__(self) -> "ASIObjectiveTurretSimulation": + return self + + def __exit__(self, *exc) -> None: + self.close() + + @property + def current_objective(self) -> Optional[str]: + return self._current_objective + + @property + def current_slot(self) -> Optional[int]: + return self._current_slot + + @property + def is_open(self) -> bool: + return self._is_open + + def _require_open(self) -> None: + if not self._is_open: + raise RuntimeError("ASI turret (simulated) is closed") + + def _retract_z_if_possible(self) -> Optional[float]: + from control._def import HOMING_ENABLED_Z, OBJECTIVE_RETRACTED_POS_MM + + if self._stage is None or not HOMING_ENABLED_Z: + return None + z_mm = self._stage.get_pos().z_mm + self._stage.move_z_to(OBJECTIVE_RETRACTED_POS_MM) + return z_mm + + def _restore_z_if_captured(self, captured_z: Optional[float]) -> None: + if captured_z is None or self._stage is None: + return + self._stage.move_z_to(captured_z) diff --git a/software/control/gui_hcs.py b/software/control/gui_hcs.py index 3b9ee1861..e7c62c224 100644 --- a/software/control/gui_hcs.py +++ b/software/control/gui_hcs.py @@ -2917,7 +2917,9 @@ def _cleanup_common(self, for_restart: bool = False): # Turret close: always release the serial port so the new process (on restart) # can acquire it, and so the motor is de-energized on full shutdown. Independent # of Z-retract success — the close path must run even if Z retract failed. - if USE_OBJECTIVE_TURRET and self.objective_changer: + # ASI turret: close() releases the serial only when the turret opened it; a + # connection shared with the Z stage is closed by self.stage.close() below. + if (USE_OBJECTIVE_TURRET or USE_ASI_OBJECTIVE_TURRET) and self.objective_changer: try: self.objective_changer.close() except Exception: diff --git a/software/control/microscope.py b/software/control/microscope.py index df501e2cb..74d838d6c 100644 --- a/software/control/microscope.py +++ b/software/control/microscope.py @@ -51,6 +51,12 @@ else: ObjectiveTurret4PosController = None +if control._def.USE_ASI_OBJECTIVE_TURRET: + from control.asi_objective_turret import ASIObjectiveTurret, ASIObjectiveTurretSimulation +else: + ASIObjectiveTurret = None + ASIObjectiveTurretSimulation = None + class ObjectiveChangerProtocol(Protocol): """Methods shared by both Xeryon and turret controllers. Controller-specific @@ -174,6 +180,28 @@ def build_from_global_config( if not objective_changer_simulated else ObjectiveTurret4PosControllerSimulation(**turret_kwargs) ) + elif control._def.USE_ASI_OBJECTIVE_TURRET: + if objective_changer_simulated: + objective_changer = ASIObjectiveTurretSimulation( + positions=control._def.ASI_OBJECTIVE_TURRET_POSITIONS, + stage=stage, + axis=str(control._def.ASI_OBJECTIVE_TURRET_AXIS_LETTER), + ) + else: + # Same MS-2000 controller as the LS50 Z stage when both are enabled: reuse its + # serial (the Z stage owns/closes it). None => turret-without-Z (or simulated + # Z): open our own connection, defaulting SN/port/baud to the ASI_Z_* values. + objective_changer = ASIObjectiveTurret( + shared_serial=squid.stage.asi.find_shared_ms2000(stage), + serial_number=_config_sn_to_str(control._def.ASI_OBJECTIVE_TURRET_SN) + or _config_sn_to_str(control._def.ASI_Z_STAGE_SN), + serial_port=(control._def.ASI_OBJECTIVE_TURRET_SERIAL_PORT or control._def.ASI_Z_SERIAL_PORT) + or None, + baudrate=int(control._def.ASI_OBJECTIVE_TURRET_BAUDRATE or control._def.ASI_Z_BAUDRATE), + axis=str(control._def.ASI_OBJECTIVE_TURRET_AXIS_LETTER), + positions=control._def.ASI_OBJECTIVE_TURRET_POSITIONS, + stage=stage, + ) camera_focus = None if control._def.SUPPORT_LASER_AUTOFOCUS: @@ -589,7 +617,8 @@ def _prepare_for_use(self, skip_init: bool = False): # Xeryon always re-homes (findIndex is fast and required). The turret skips # homing on a software restart: the controller retains its position counter # across close()/re-init (the motor is de-energized but the drive stays - # powered), so a re-home would just be wasted motion. + # powered), so a re-home would just be wasted motion. The ASI turret has no + # homing at all: its home() only refreshes the tracked slot, never moves. if control._def.USE_XERYON or not skip_init: self.addons.objective_changer.home() if control._def.USE_XERYON: diff --git a/software/tests/control/test_asi_objective_turret.py b/software/tests/control/test_asi_objective_turret.py new file mode 100644 index 000000000..4bd4c513e --- /dev/null +++ b/software/tests/control/test_asi_objective_turret.py @@ -0,0 +1,277 @@ +import pytest + +import control._def +import control.asi_objective_turret as aot +from squid.stage.asi import MS2000Serial + +POSITIONS = {"2x": 1, "4x": 2, "10x": 3, "20x": 4, "40x": 5, "60x": 6} + + +class _FakeSerialConn: + """Scripted pyserial-like object: records writes, pops queued replies, tracks close.""" + + def __init__(self, replies=(), default=b""): + self.written = [] + self.replies = list(replies) + self.default = default + self.closed = False + + def write(self, data): + self.written.append(data) + + def read_until(self, expected=b"\n"): + return self.replies.pop(0) if self.replies else self.default + + def close(self): + self.closed = True + + +class FakeStage: + """Records move_z_to calls and reports a preset Z position.""" + + def __init__(self, z_mm: float = 3.5): + self._z_mm = z_mm + self.z_moves: list[float] = [] + + def move_z_to(self, abs_mm: float, blocking: bool = True): + self.z_moves.append(abs_mm) + self._z_mm = abs_mm + + def get_pos(self): + class _Pos: + pass + + p = _Pos() + p.z_mm = self._z_mm + return p + + +def _shared_turret(replies, stage=None, positions=POSITIONS, **kwargs): + """Turret on a shared (not owned) scripted transport; first reply feeds the ctor W T probe.""" + conn = _FakeSerialConn(replies=replies) + turret = aot.ASIObjectiveTurret(shared_serial=MS2000Serial(conn), positions=positions, stage=stage, **kwargs) + return turret, conn + + +# --- simulation --------------------------------------------------------------- + + +def test_sim_init_open(): + sim = aot.ASIObjectiveTurretSimulation(positions=POSITIONS) + assert sim.is_open is True + assert sim.current_objective is None + assert sim.current_slot is None # unknown until commanded + + +@pytest.mark.parametrize("name", sorted(POSITIONS)) +def test_sim_move_to_each_known_objective(name): + sim = aot.ASIObjectiveTurretSimulation(positions=POSITIONS) + sim.move_to_objective(name) + assert sim.current_objective == name + assert sim.current_slot == POSITIONS[name] + + +def test_sim_unknown_objective_raises_key_error(): + sim = aot.ASIObjectiveTurretSimulation(positions=POSITIONS) + with pytest.raises(KeyError, match="Valid names"): + sim.move_to_objective("95x") + + +def test_sim_home_is_motionless(monkeypatch): + # The ASI turret has no homing; home() must not move the turret or Z. + monkeypatch.setattr(control._def, "HOMING_ENABLED_Z", True, raising=False) + stage = FakeStage() + sim = aot.ASIObjectiveTurretSimulation(positions=POSITIONS, stage=stage) + sim.move_to_objective("10x") + sim.home() + assert sim.current_objective is None # tracked name cleared + assert sim.current_slot == 3 # slot unchanged: no rotation happened + assert stage.z_moves == [control._def.OBJECTIVE_RETRACTED_POS_MM, 3.5] # only the earlier move's dance + + +def test_sim_retract_and_restore_z(monkeypatch): + monkeypatch.setattr(control._def, "HOMING_ENABLED_Z", True, raising=False) + stage = FakeStage(z_mm=3.5) + sim = aot.ASIObjectiveTurretSimulation(positions=POSITIONS, stage=stage) + sim.move_to_objective("4x") + assert stage.z_moves == [control._def.OBJECTIVE_RETRACTED_POS_MM, 3.5] + + +def test_sim_skips_z_without_stage(monkeypatch): + monkeypatch.setattr(control._def, "HOMING_ENABLED_Z", True, raising=False) + sim = aot.ASIObjectiveTurretSimulation(positions=POSITIONS, stage=None) + sim.move_to_objective("4x") # must not raise + + +def test_sim_skips_z_when_homing_z_disabled(monkeypatch): + monkeypatch.setattr(control._def, "HOMING_ENABLED_Z", False, raising=False) + stage = FakeStage() + sim = aot.ASIObjectiveTurretSimulation(positions=POSITIONS, stage=stage) + sim.move_to_objective("4x") + assert stage.z_moves == [] + + +def test_sim_alias_same_slot_skips_dance(monkeypatch): + monkeypatch.setattr(control._def, "HOMING_ENABLED_Z", True, raising=False) + stage = FakeStage() + positions = {"4x": 1, "4x-phase": 1, "10x": 2} + sim = aot.ASIObjectiveTurretSimulation(positions=positions, stage=stage) + sim.move_to_objective("4x") + moves_after_first = list(stage.z_moves) + sim.move_to_objective("4x-phase") # same slot: tracked-name-only no-op + assert sim.current_objective == "4x-phase" + assert stage.z_moves == moves_after_first + + +def test_sim_restore_z_false_skips_restore(monkeypatch): + monkeypatch.setattr(control._def, "HOMING_ENABLED_Z", True, raising=False) + stage = FakeStage() + sim = aot.ASIObjectiveTurretSimulation(positions=POSITIONS, stage=stage) + sim.move_to_objective("4x", restore_z=False) + assert stage.z_moves == [control._def.OBJECTIVE_RETRACTED_POS_MM] + + +def test_sim_ops_after_close_raise(): + sim = aot.ASIObjectiveTurretSimulation(positions=POSITIONS) + sim.close() + with pytest.raises(RuntimeError): + sim.move_to_objective("4x") + + +def test_sim_close_idempotent_and_context_manager(): + with aot.ASIObjectiveTurretSimulation(positions=POSITIONS) as sim: + assert sim.is_open + assert not sim.is_open + sim.close() # second close must not raise + + +def test_invalid_slot_positions_raise(): + for bad in ({"4x": 0}, {"4x": 7}, {"4x": "one"}): + with pytest.raises(ValueError): + aot.ASIObjectiveTurretSimulation(positions=bad) + with pytest.raises(ValueError): + aot.ASIObjectiveTurret(shared_serial=MS2000Serial(_FakeSerialConn()), positions=bad) + + +# --- real controller over a scripted transport -------------------------------- + + +def test_move_frames_raw_slot_command(): + # Slot index is sent RAW ('M T=3'), not scaled by the 0.1 um unit factor. + turret, conn = _shared_turret([b":A 1\r\n", b":A\r\n", b"N\r\n", b":A 3\r\n"]) + assert turret.current_slot == 1 # seeded from the ctor W T probe + turret.move_to_objective("10x") + assert conn.written == [b"W T\r", b"M T=3\r", b"/\r", b"W T\r"] + assert turret.current_objective == "10x" + assert turret.current_slot == 3 + + +def test_wait_idle_polls_global_busy(): + turret, conn = _shared_turret([b":A 1\r\n", b":A\r\n", b"B\r\n", b"B\r\n", b"N\r\n", b":A 2\r\n"]) + turret.move_to_objective("4x") + assert conn.written.count(b"/\r") == 3 + + +def test_move_timeout_raises_and_still_restores_z(monkeypatch): + monkeypatch.setattr(control._def, "HOMING_ENABLED_Z", True, raising=False) + stage = FakeStage(z_mm=2.0) + conn = _FakeSerialConn(replies=[b":A 1\r\n"], default=b"B\r\n") # never idle + turret = aot.ASIObjectiveTurret(shared_serial=MS2000Serial(conn), positions=POSITIONS, stage=stage) + with pytest.raises(RuntimeError, match="idle"): + turret.move_to_objective("4x", timeout_s=0.12) + assert stage.z_moves == [control._def.OBJECTIVE_RETRACTED_POS_MM, 2.0] # finally restored + + +def test_probe_seed_short_circuits_move(): + # Ctor W T said slot 4; moving to the slot-4 objective must not rotate. + turret, conn = _shared_turret([b":A 4\r\n"]) + turret.move_to_objective("20x") + assert conn.written == [b"W T\r"] # no M command + assert turret.current_objective == "20x" + + +def test_probe_garbage_means_unknown_slot(): + turret, conn = _shared_turret([b":N-1\r\n", b":A\r\n", b"N\r\n"]) + assert turret.current_slot is None + turret.move_to_objective("2x") # unknown never short-circuits: rotation commanded + assert b"M T=1\r" in conn.written + + +def test_unknown_objective_keyerror_real(): + turret, _ = _shared_turret([b":A 1\r\n"]) + with pytest.raises(KeyError, match="Valid names"): + turret.move_to_objective("95x") + + +def test_shared_serial_not_closed_on_close(): + turret, conn = _shared_turret([b":A 1\r\n"]) + assert turret.owns_serial is False + turret.close() + assert conn.closed is False # the Z stage owns the shared transport + + +def test_owned_serial_closed_on_close(monkeypatch): + conn = _FakeSerialConn(replies=[b":A 1\r\n"]) + monkeypatch.setattr(MS2000Serial, "open", classmethod(lambda cls, *a, **k: MS2000Serial(conn))) + turret = aot.ASIObjectiveTurret(serial_port="/dev/FAKE", positions=POSITIONS) + assert turret.owns_serial is True + turret.close() + assert conn.closed is True + + +def test_owned_ctor_probe_failure_closes_serial(monkeypatch): + conn = _FakeSerialConn(replies=[]) # dead air, both probe attempts + monkeypatch.setattr(MS2000Serial, "open", classmethod(lambda cls, *a, **k: MS2000Serial(conn))) + with pytest.raises(RuntimeError): + aot.ASIObjectiveTurret(serial_port="/dev/FAKE", positions=POSITIONS) + assert conn.closed is True # no leaked handle + + +def test_home_is_motionless_real(): + turret, conn = _shared_turret([b":A 2\r\n", b":A 2\r\n"]) + turret.home() + assert all(not w.startswith(b"M ") for w in conn.written) # W T probes only, no motion + assert turret.current_slot == 2 + + +def test_error_ack_on_move_raises(): + turret, _ = _shared_turret([b":A 1\r\n", b":N-4\r\n"]) + with pytest.raises(RuntimeError, match="N-4"): + turret.move_to_objective("4x") + + +def test_turret_without_port_or_sn_raises(): + with pytest.raises(RuntimeError, match="ASI_OBJECTIVE_TURRET"): + aot.ASIObjectiveTurret(positions=POSITIONS) + + +# --- simulated end-to-end through Microscope ----------------------------------- + + +def _build_asi_scope(monkeypatch, skip_init): + import control.microscope + from control.asi_objective_turret import ASIObjectiveTurret, ASIObjectiveTurretSimulation + + monkeypatch.setattr(control._def, "USE_ASI_Z_STAGE", True, raising=False) + monkeypatch.setattr(control._def, "SIMULATE_ASI_Z_STAGE", True, raising=False) + monkeypatch.setattr(control._def, "USE_ASI_OBJECTIVE_TURRET", True, raising=False) + monkeypatch.setattr(control._def, "SIMULATE_OBJECTIVE_CHANGER", True, raising=False) + # The conditional import left these None (flag was off at import time). + monkeypatch.setattr(control.microscope, "ASIObjectiveTurret", ASIObjectiveTurret, raising=False) + monkeypatch.setattr(control.microscope, "ASIObjectiveTurretSimulation", ASIObjectiveTurretSimulation, raising=False) + return control.microscope.Microscope.build_from_global_config(simulated=True, skip_init=skip_init) + + +def test_microscope_builds_asi_turret_simulated(monkeypatch): + scope = _build_asi_scope(monkeypatch, skip_init=True) + changer = scope.addons.objective_changer + assert isinstance(changer, aot.ASIObjectiveTurretSimulation) + changer.move_to_objective(control._def.DEFAULT_OBJECTIVE) + assert changer.current_objective == control._def.DEFAULT_OBJECTIVE + scope.close() + + +def test_microscope_startup_selects_default_objective(monkeypatch): + scope = _build_asi_scope(monkeypatch, skip_init=False) + assert scope.addons.objective_changer.current_objective == control._def.DEFAULT_OBJECTIVE + scope.close() diff --git a/software/tests/control/test_objective_changer_config.py b/software/tests/control/test_objective_changer_config.py index 452e1aefa..e4424cd23 100644 --- a/software/tests/control/test_objective_changer_config.py +++ b/software/tests/control/test_objective_changer_config.py @@ -25,3 +25,27 @@ def test_mutual_exclusion_allows_neither(): def test_objective_turret_positions_shape(): assert len(OBJECTIVE_TURRET_POSITIONS) == 4 assert sorted(OBJECTIVE_TURRET_POSITIONS.values()) == [1, 2, 3, 4] + + +@pytest.mark.parametrize( + "xeryon,turret,asi", + [(True, True, False), (True, False, True), (False, True, True), (True, True, True)], +) +def test_mutual_exclusion_any_two_of_three_raise(xeryon, turret, asi): + with pytest.raises(ValueError, match="mutually exclusive"): + _validate_objective_changer_flags(xeryon, turret, asi) + + +def test_mutual_exclusion_allows_asi_turret_only(): + _validate_objective_changer_flags(use_xeryon=False, use_turret=False, use_asi_turret=True) + + +def test_asi_turret_positions_shape(): + from control._def import ASI_OBJECTIVE_TURRET_POSITIONS, OBJECTIVES, DEFAULT_OBJECTIVE + + assert len(ASI_OBJECTIVE_TURRET_POSITIONS) == 6 + assert sorted(ASI_OBJECTIVE_TURRET_POSITIONS.values()) == [1, 2, 3, 4, 5, 6] + # Keys must be real objective names so the GUI dropdown -> move_to_objective flow works, + # and the default dict must cover the startup DEFAULT_OBJECTIVE fallback. + assert set(ASI_OBJECTIVE_TURRET_POSITIONS) <= set(OBJECTIVES) + assert "20x" in ASI_OBJECTIVE_TURRET_POSITIONS From 95924d4556eeb468bbf5110c1fbecbad2fe79520 Mon Sep 17 00:00:00 2001 From: You Yan Date: Fri, 10 Jul 2026 14:25:39 -0700 Subject: [PATCH 3/9] feat(tools): --turret probe/move mode in asi_z_bringup --turret: read-only raw 'W ' + '/' replies (answers the W-T-semantics question). --turret-slot N (behind --allow-motion + confirmation): rotate, log '/' busy transitions with timestamps (measures whether the global busy byte covers turret rotation), read back, repeat same slot. read_only_checks switched to the public backend.serial accessor. Co-Authored-By: Claude Fable 5 --- software/tools/asi_z_bringup.py | 37 ++++++++++++++++++++++++++++++++- 1 file changed, 36 insertions(+), 1 deletion(-) diff --git a/software/tools/asi_z_bringup.py b/software/tools/asi_z_bringup.py index 598496ec9..2178acc82 100644 --- a/software/tools/asi_z_bringup.py +++ b/software/tools/asi_z_bringup.py @@ -95,7 +95,7 @@ def read_only_checks(backend): report_position(backend, "Position") log.info(f"Busy: {backend.is_moving()} (expect False on an idle stage)") if isinstance(backend, LS50Controller): - with_serial: MS2000Serial = backend._serial + with_serial: MS2000Serial = backend.serial for cmd in ("BU", "N"): # build info / who try: log.info(f"{cmd!r} -> {with_serial.command(cmd)!r}") @@ -139,6 +139,34 @@ def fence_test(backend, fence_mm: float): report_position(backend, "Back at start") +def turret_probe(serial, axis: str): + """Read-only: raw 'W ' and '/' replies, verbatim (answers the W-T-semantics question).""" + log.info(f"--- Turret probe (axis {axis!r}) ---") + for cmd in (f"W {axis}", "/"): + reply = serial.command(cmd, check_error=False) + log.info(f"{cmd!r} -> {reply!r}") + + +def turret_move_test(serial, axis: str, slot: int): + """Rotate to a slot, logging '/' busy transitions with timestamps, then read back.""" + log.info(f"--- Turret move test: slot {slot} ---") + log.info(f"'M {axis}={slot}' -> {serial.command(f'M {axis}={slot}', check_error=False)!r}") + t0 = time.monotonic() + last = None + deadline = t0 + 30.0 + while time.monotonic() < deadline: + busy = "B" in serial.command("/").upper() + if busy != last: + log.info(f"t={time.monotonic() - t0:6.2f}s busy={busy}") + last = busy + if last is False: + break + time.sleep(0.05) + log.info(f"readback 'W {axis}' -> {serial.command(f'W {axis}', check_error=False)!r}") + log.info(f"same-slot repeat 'M {axis}={slot}' -> {serial.command(f'M {axis}={slot}', check_error=False)!r}") + log.info("If busy never went True, '/' may not cover turret rotation (note for the driver).") + + def main(): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--sn", help="Controller USB serial number (resolved to a port)") @@ -150,6 +178,9 @@ def main(): parser.add_argument("--jog-mm", type=float, default=0.5, help="Jog distance in native mm (default 0.5, away)") parser.add_argument("--fence-mm", type=float, default=0.0, help="Also test the +/- soft-limit fence (0 = skip)") parser.add_argument("--scan-bauds", action="store_true", help="Try each ASI baud rate and report which answers") + parser.add_argument("--turret", action="store_true", help="Probe the objective turret axis (read-only)") + parser.add_argument("--turret-axis", default="T", help="Turret axis letter (default T)") + parser.add_argument("--turret-slot", type=int, help="Rotate to this slot (1..6); requires --allow-motion") args = parser.parse_args() if args.scan_bauds: @@ -167,6 +198,8 @@ def main(): backend = connect(args) try: read_only_checks(backend) + if args.turret and isinstance(backend, LS50Controller): + turret_probe(backend.serial, args.turret_axis) if not args.allow_motion: log.info("Read-only run complete. Re-run with --allow-motion for the jog test.") @@ -181,6 +214,8 @@ def main(): jog_test(backend, args.jog_mm) if args.fence_mm > 0: fence_test(backend, args.fence_mm) + if args.turret and args.turret_slot and isinstance(backend, LS50Controller): + turret_move_test(backend.serial, args.turret_axis, args.turret_slot) log.info("--- Done. Next steps for the machine ini ---") log.info("[GENERAL]: use_asi_z_stage = True, asi_z_stage_sn = , asi_z_travel_mm = 50") From 4587873bd6ca4883256a2a99abbca429bc3ba59e Mon Sep 17 00:00:00 2001 From: You Yan Date: Fri, 10 Jul 2026 15:37:59 -0700 Subject: [PATCH 4/9] refactor(stage): move CombinedStage to vendor-neutral squid/stage/composite.py The composite went multi-vendor (PI V-308, ASI LS50) but lived in the vendor-named pi.py, forcing the ASI turret's find_shared_ms2000 to duck-type around an import cycle and microscope.py's ASI branch to import through the PI module. CombinedStage now lives in squid/stage/composite.py; pi.py keeps a compatibility re-export; find_shared_ms2000 uses a real isinstance; canonical import sites (microscope.py, tests) migrated. Co-Authored-By: Claude Fable 5 --- software/control/microscope.py | 3 +- software/squid/stage/asi.py | 18 +++-- software/squid/stage/composite.py | 123 +++++++++++++++++++++++++++++ software/squid/stage/pi.py | 109 +------------------------ software/tests/squid/test_stage.py | 9 ++- 5 files changed, 142 insertions(+), 120 deletions(-) create mode 100644 software/squid/stage/composite.py diff --git a/software/control/microscope.py b/software/control/microscope.py index 74d838d6c..dcd87a115 100644 --- a/software/control/microscope.py +++ b/software/control/microscope.py @@ -30,6 +30,7 @@ import squid.logging import squid.stage.asi import squid.stage.cephla +import squid.stage.composite import squid.stage.pi import squid.stage.utils @@ -407,7 +408,7 @@ def build_from_global_config(simulated: bool = False, skip_init: bool = False) - stage_config=stage_config, ) if z_stage is not None: - stage = squid.stage.pi.CombinedStage(xy_stage=stage, z_stage=z_stage, stage_config=stage_config) + stage = squid.stage.composite.CombinedStage(xy_stage=stage, z_stage=z_stage, stage_config=stage_config) addons = MicroscopeAddons.build_from_global_config( stage, low_level_devices.microcontroller, simulated=simulated, skip_init=skip_init diff --git a/software/squid/stage/asi.py b/software/squid/stage/asi.py index e78a726fb..686f3d482 100644 --- a/software/squid/stage/asi.py +++ b/software/squid/stage/asi.py @@ -4,7 +4,7 @@ Provides (1) ``MS2000Serial``, the CR-terminated ASI command transport; (2) ``LS50Controller``, the single-axis driver; (3) ``_SimulatedLS50``, a pure-Python stand-in for hardware-free / CI use; and (4) ``ASIZStage``, the Z-only ``squid.abc.AbstractStage`` adapter (pair with an XY -stage via ``squid.stage.pi.CombinedStage``). +stage via ``squid.stage.composite.CombinedStage``). Frame and units: the controller's native unit is 1/10 um (10000 per mm). Native 0 is the power-on position, which by convention is the RETRACTED end; native positive is away from the @@ -396,14 +396,16 @@ def _no_xy(self, name: str): def find_shared_ms2000(stage) -> Optional[MS2000Serial]: """Return the MS2000Serial of an ASI Z stage embedded in ``stage``, else None. - Accepts an ASIZStage directly or a composite exposing a ``z_stage`` property (duck-typed - to avoid an import cycle with squid.stage.pi). Returns None for simulated backends and - non-ASI stages. Ownership stays with the Z stage: callers must NOT close the returned - transport -- it is released by stage.close(). + Accepts an ASIZStage directly or a CombinedStage wrapping one. Returns None for + simulated backends and non-ASI stages. Ownership stays with the Z stage: callers must + NOT close the returned transport -- it is released by stage.close(). """ - for candidate in (stage, getattr(stage, "z_stage", None)): - if isinstance(candidate, ASIZStage): - return candidate.ms2000_serial + from squid.stage.composite import CombinedStage # here to keep module import light + + if isinstance(stage, CombinedStage): + stage = stage.z_stage + if isinstance(stage, ASIZStage): + return stage.ms2000_serial return None diff --git a/software/squid/stage/composite.py b/software/squid/stage/composite.py new file mode 100644 index 000000000..abe06aeaf --- /dev/null +++ b/software/squid/stage/composite.py @@ -0,0 +1,123 @@ +"""Vendor-neutral composite stage: an XY stage plus a Z-only focus stage as one AbstractStage. + +``CombinedStage`` routes X / Y / theta to the wrapped XY stage (Cephla or Prior) and Z to a +Z-only external focus stage (e.g. the PI V-308 in ``squid.stage.pi`` or the ASI LS50 in +``squid.stage.asi``). Consumers keep programming against the single ``AbstractStage`` +interface; the informal contract a z_stage must satisfy is documented on the class. +""" + +from __future__ import annotations + +from typing import Optional + +import squid.logging +from squid.abc import AbstractStage, Pos, StageStage +from squid.config import StageConfig + +_log = squid.logging.get_logger(__name__) + + +class CombinedStage(AbstractStage): + """AbstractStage routing X / Y / theta to xy_stage and Z to z_stage (an external Z-only focus stage, e.g. the V-308 or the ASI LS50).""" + + def __init__(self, xy_stage: AbstractStage, z_stage: AbstractStage, stage_config: Optional[StageConfig] = None): + super().__init__(stage_config or xy_stage.get_config()) + self._xy = xy_stage + self._z = z_stage + self._scanning_position_z_mm = None # set/read by squid.stage.utils loading/scanning flow + + # The GUI snaps Z step sizes through get_config().Z_AXIS (AutoFocus / multipoint) and via + # z_mm_to_usteps (navigation). Present a Z axis whose resolution is the Z stage's own + # (continuous ~10 nm) grid instead of the wrapped XY stepper grid, so Z-stack/autofocus + # steps are not snapped to the coarse stepper microstep grid. Only the resolution fields + # are overridden; range/speed/sign are preserved. + z_usteps_per_mm = abs(self._z.z_mm_to_usteps(1.0)) if hasattr(self._z, "z_mm_to_usteps") else 0.0 + if z_usteps_per_mm: + fine_z = self._config.Z_AXIS.model_copy( + update={"SCREW_PITCH": 1.0, "MICROSTEPS_PER_STEP": 1, "FULL_STEPS_PER_REV": float(z_usteps_per_mm)} + ) + self._config = self._config.model_copy(update={"Z_AXIS": fine_z}) + + @property + def z_stage(self) -> AbstractStage: + """The wrapped Z-only stage (public: addons sharing its transport walk through here).""" + return self._z + + def move_x(self, rel_mm: float, blocking: bool = True): + self._xy.move_x(rel_mm, blocking) + + def move_y(self, rel_mm: float, blocking: bool = True): + self._xy.move_y(rel_mm, blocking) + + def move_z(self, rel_mm: float, blocking: bool = True): + self._z.move_z(rel_mm, blocking) + + def move_x_to(self, abs_mm: float, blocking: bool = True): + self._xy.move_x_to(abs_mm, blocking) + + def move_y_to(self, abs_mm: float, blocking: bool = True): + self._xy.move_y_to(abs_mm, blocking) + + def move_z_to(self, abs_mm: float, blocking: bool = True): + self._z.move_z_to(abs_mm, blocking) + + def get_pos(self) -> Pos: + xy, z = self._xy.get_pos(), self._z.get_pos() + return Pos(x_mm=xy.x_mm, y_mm=xy.y_mm, z_mm=z.z_mm, theta_rad=xy.theta_rad) + + def get_state(self) -> StageStage: + return StageStage(busy=self._xy.get_state().busy or self._z.get_state().busy) + + def is_referenced(self) -> bool: + return self._z.is_referenced() + + def home(self, x: bool, y: bool, z: bool, theta: bool, blocking: bool = True): + xy_requested = x or y or theta + if z: + # Z must finish retracting before any XY sweep (e.g. the V-308 voice coil is not self-locking). + self._z.home(False, False, True, False, blocking or xy_requested) + if xy_requested: + self._xy.home(x, y, False, theta, blocking) + + def zero(self, x: bool, y: bool, z: bool, theta: bool, blocking: bool = True): + if x or y or theta: + self._xy.zero(x, y, False, theta, blocking) + if z: + self._z.zero(False, False, True, False, blocking) + + def set_limits( + self, + x_pos_mm: Optional[float] = None, + x_neg_mm: Optional[float] = None, + y_pos_mm: Optional[float] = None, + y_neg_mm: Optional[float] = None, + z_pos_mm: Optional[float] = None, + z_neg_mm: Optional[float] = None, + theta_pos_rad: Optional[float] = None, + theta_neg_rad: Optional[float] = None, + ): + self._xy.set_limits( + x_pos_mm=x_pos_mm, + x_neg_mm=x_neg_mm, + y_pos_mm=y_pos_mm, + y_neg_mm=y_neg_mm, + theta_pos_rad=theta_pos_rad, + theta_neg_rad=theta_neg_rad, + ) + self._z.set_limits(z_pos_mm=z_pos_mm, z_neg_mm=z_neg_mm) + + # The GUI (NavigationWidget.set_deltaX/Y/Z) calls these stepper-style helpers on the stage, so + # the wrapper must expose them. X/Y come from the wrapped XY stage; Z comes from the external + # Z stage's own grid, not the XY stepper grid. + def x_mm_to_usteps(self, mm: float): + return self._xy.x_mm_to_usteps(mm) + + def y_mm_to_usteps(self, mm: float): + return self._xy.y_mm_to_usteps(mm) + + def z_mm_to_usteps(self, mm: float): + return self._z.z_mm_to_usteps(mm) + + def close(self): + self._z.close() # the external Z stage's serial handle; Cephla/Prior XY close() is a no-op + self._xy.close() diff --git a/software/squid/stage/pi.py b/software/squid/stage/pi.py index 861e3a540..4e5673ae0 100644 --- a/software/squid/stage/pi.py +++ b/software/squid/stage/pi.py @@ -5,7 +5,7 @@ pipython's pure-Python PISerial gateway (no proprietary PI GCS DLL required), with ``pipython`` imported lazily so this module imports fine without it; (2) ``_SimulatedC414``, a pure-Python stand-in for hardware-free / CI use; and (3) two ``squid.abc.AbstractStage`` -adapters -- ``PIFocusStage`` (Z-only) and ``CombinedStage`` (XY delegate + V-308 Z). +adapter ``PIFocusStage`` (Z-only; pair with an XY stage via ``squid.stage.composite.CombinedStage``). Z is pure pass-through mm: the controller's native absolute mm is Squid's Z mm, with no sign flip or offset and no use of ``Z_AXIS.MOVEMENT_SIGN`` (that is a Cephla-stepper calibration). @@ -21,6 +21,7 @@ import squid.logging from squid.abc import AbstractStage, Pos, StageStage from squid.config import StageConfig +from squid.stage.composite import CombinedStage # re-export: CombinedStage lived here until it went multi-vendor _log = squid.logging.get_logger(__name__) @@ -277,112 +278,6 @@ def _no_xy(self, name: str): self._log.warning(f"{name} ignored: PIFocusStage is a Z-only focus drive (pair via CombinedStage).") -class CombinedStage(AbstractStage): - """AbstractStage routing X / Y / theta to xy_stage and Z to z_stage (an external Z-only focus stage, e.g. the V-308 or the ASI LS50).""" - - def __init__(self, xy_stage: AbstractStage, z_stage: AbstractStage, stage_config: Optional[StageConfig] = None): - super().__init__(stage_config or xy_stage.get_config()) - self._xy = xy_stage - self._z = z_stage - self._scanning_position_z_mm = None # set/read by squid.stage.utils loading/scanning flow - - # The GUI snaps Z step sizes through get_config().Z_AXIS (AutoFocus / multipoint) and via - # z_mm_to_usteps (navigation). Present a Z axis whose resolution is the Z stage's own - # (continuous ~10 nm) grid instead of the wrapped XY stepper grid, so Z-stack/autofocus - # steps are not snapped to the coarse stepper microstep grid. Only the resolution fields - # are overridden; range/speed/sign are preserved. - z_usteps_per_mm = abs(self._z.z_mm_to_usteps(1.0)) if hasattr(self._z, "z_mm_to_usteps") else 0.0 - if z_usteps_per_mm: - fine_z = self._config.Z_AXIS.model_copy( - update={"SCREW_PITCH": 1.0, "MICROSTEPS_PER_STEP": 1, "FULL_STEPS_PER_REV": float(z_usteps_per_mm)} - ) - self._config = self._config.model_copy(update={"Z_AXIS": fine_z}) - - @property - def z_stage(self) -> AbstractStage: - """The wrapped Z-only stage (public: addons sharing its transport walk through here).""" - return self._z - - def move_x(self, rel_mm: float, blocking: bool = True): - self._xy.move_x(rel_mm, blocking) - - def move_y(self, rel_mm: float, blocking: bool = True): - self._xy.move_y(rel_mm, blocking) - - def move_z(self, rel_mm: float, blocking: bool = True): - self._z.move_z(rel_mm, blocking) - - def move_x_to(self, abs_mm: float, blocking: bool = True): - self._xy.move_x_to(abs_mm, blocking) - - def move_y_to(self, abs_mm: float, blocking: bool = True): - self._xy.move_y_to(abs_mm, blocking) - - def move_z_to(self, abs_mm: float, blocking: bool = True): - self._z.move_z_to(abs_mm, blocking) - - def get_pos(self) -> Pos: - xy, z = self._xy.get_pos(), self._z.get_pos() - return Pos(x_mm=xy.x_mm, y_mm=xy.y_mm, z_mm=z.z_mm, theta_rad=xy.theta_rad) - - def get_state(self) -> StageStage: - return StageStage(busy=self._xy.get_state().busy or self._z.get_state().busy) - - def is_referenced(self) -> bool: - return self._z.is_referenced() - - def home(self, x: bool, y: bool, z: bool, theta: bool, blocking: bool = True): - xy_requested = x or y or theta - if z: - # Z must finish retracting before any XY sweep (e.g. the V-308 voice coil is not self-locking). - self._z.home(False, False, True, False, blocking or xy_requested) - if xy_requested: - self._xy.home(x, y, False, theta, blocking) - - def zero(self, x: bool, y: bool, z: bool, theta: bool, blocking: bool = True): - if x or y or theta: - self._xy.zero(x, y, False, theta, blocking) - if z: - self._z.zero(False, False, True, False, blocking) - - def set_limits( - self, - x_pos_mm: Optional[float] = None, - x_neg_mm: Optional[float] = None, - y_pos_mm: Optional[float] = None, - y_neg_mm: Optional[float] = None, - z_pos_mm: Optional[float] = None, - z_neg_mm: Optional[float] = None, - theta_pos_rad: Optional[float] = None, - theta_neg_rad: Optional[float] = None, - ): - self._xy.set_limits( - x_pos_mm=x_pos_mm, - x_neg_mm=x_neg_mm, - y_pos_mm=y_pos_mm, - y_neg_mm=y_neg_mm, - theta_pos_rad=theta_pos_rad, - theta_neg_rad=theta_neg_rad, - ) - self._z.set_limits(z_pos_mm=z_pos_mm, z_neg_mm=z_neg_mm) - - # The GUI (NavigationWidget.set_deltaX/Y/Z) calls these stepper-style helpers on the stage, so - # the wrapper must expose them. X/Y come from the wrapped XY stage; Z comes from the external - # Z stage's own grid, not the XY stepper grid. - def x_mm_to_usteps(self, mm: float): - return self._xy.x_mm_to_usteps(mm) - - def y_mm_to_usteps(self, mm: float): - return self._xy.y_mm_to_usteps(mm) - - def z_mm_to_usteps(self, mm: float): - return self._z.z_mm_to_usteps(mm) - - def close(self): - self._z.close() # the external Z stage's serial handle; Cephla/Prior XY close() is a no-op - self._xy.close() - - class C414FocusStage: """Single-axis closed-loop focus drive (V-308 on a C-414), GCS 2.0 via pipython. diff --git a/software/tests/squid/test_stage.py b/software/tests/squid/test_stage.py index 2605e073a..6a1de8a8d 100644 --- a/software/tests/squid/test_stage.py +++ b/software/tests/squid/test_stage.py @@ -5,6 +5,7 @@ import squid.stage.prior import squid.stage.utils import squid.stage.pi +import squid.stage.composite import squid.stage.asi import squid.config import squid.abc @@ -66,7 +67,7 @@ def _sim_combined_stage(z_stage=None): """Simulated Cephla XY + Z-only composite (PI Z by default); returns (combined, xy, z).""" xy = squid.stage.cephla.CephlaStage(get_test_micro(), squid.config.get_stage_config()) z = z_stage if z_stage is not None else _sim_pi_stage() - combined = squid.stage.pi.CombinedStage(xy_stage=xy, z_stage=z, stage_config=squid.config.get_stage_config()) + combined = squid.stage.composite.CombinedStage(xy_stage=xy, z_stage=z, stage_config=squid.config.get_stage_config()) return combined, xy, z @@ -167,7 +168,7 @@ def test_microscope_wraps_pi_focus_when_enabled(monkeypatch): monkeypatch.setattr(control._def, "USE_PI_FOCUS_STAGE", True, raising=False) monkeypatch.setattr(control._def, "SIMULATE_PI_FOCUS_STAGE", True, raising=False) scope = control.microscope.Microscope.build_from_global_config(simulated=True, skip_init=True) - assert isinstance(scope.stage, squid.stage.pi.CombinedStage) + assert isinstance(scope.stage, squid.stage.composite.CombinedStage) # skip_init leaves the V-308 unreferenced (reference=...and not skip_init); reference before moving. scope.stage.home(x=False, y=False, z=True, theta=False) scope.stage.move_z_to(0.3) @@ -687,7 +688,7 @@ def test_microscope_wraps_asi_z_when_enabled(monkeypatch): monkeypatch.setattr(control._def, "USE_ASI_Z_STAGE", True, raising=False) monkeypatch.setattr(control._def, "SIMULATE_ASI_Z_STAGE", True, raising=False) scope = control.microscope.Microscope.build_from_global_config(simulated=True, skip_init=True) - assert isinstance(scope.stage, squid.stage.pi.CombinedStage) + assert isinstance(scope.stage, squid.stage.composite.CombinedStage) scope.stage.move_z_to(0.3) assert abs(scope.stage.get_pos().z_mm - 0.3) < 1e-9 scope.close() # exercises Microscope.close() -> CombinedStage.close() -> ASIZStage.close() @@ -716,7 +717,7 @@ def test_asi_home_xyz_retracts_z_before_xy(monkeypatch): monkeypatch.setattr(control._def, "HOMING_ENABLED_Y", False, raising=False) scope = control.microscope.Microscope.build_from_global_config(simulated=True, skip_init=True) - assert isinstance(scope.stage, squid.stage.pi.CombinedStage) # ASI Z wrapped, not stepper Z + assert isinstance(scope.stage, squid.stage.composite.CombinedStage) # ASI Z wrapped, not stepper Z scope.stage.move_z_to(2.0) scope.home_xyz() # The external-Z branch retracts to the home target (squid 0 = native 0, the retracted From 091603784bef96e6a866fba5a3c158c789a279fb Mon Sep 17 00:00:00 2001 From: You Yan Date: Fri, 10 Jul 2026 19:00:36 -0700 Subject: [PATCH 5/9] test(turret): e2e scope sets ASI_Z_TRAVEL_MM (find-zero requires it now) Co-Authored-By: Claude Fable 5 --- software/tests/control/test_asi_objective_turret.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/software/tests/control/test_asi_objective_turret.py b/software/tests/control/test_asi_objective_turret.py index 4bd4c513e..aae48a3ee 100644 --- a/software/tests/control/test_asi_objective_turret.py +++ b/software/tests/control/test_asi_objective_turret.py @@ -254,6 +254,9 @@ def _build_asi_scope(monkeypatch, skip_init): monkeypatch.setattr(control._def, "USE_ASI_Z_STAGE", True, raising=False) monkeypatch.setattr(control._def, "SIMULATE_ASI_Z_STAGE", True, raising=False) + # find-zero (default on) derives its overdrive from the physical travel; a configured + # machine always sets this. + monkeypatch.setattr(control._def, "ASI_Z_TRAVEL_MM", 50.0, raising=False) monkeypatch.setattr(control._def, "USE_ASI_OBJECTIVE_TURRET", True, raising=False) monkeypatch.setattr(control._def, "SIMULATE_OBJECTIVE_CHANGER", True, raising=False) # The conditional import left these None (flag was off at import time). From 6a636e68ea18f076c8e71941472e9d3a3a7b835d Mon Sep 17 00:00:00 2001 From: You Yan Date: Sat, 11 Jul 2026 10:43:25 -0700 Subject: [PATCH 6/9] =?UTF-8?q?refactor(turret):=20simplification=20pass?= =?UTF-8?q?=20=E2=80=94=20shared=20Z-safety=20helpers,=20protocol=20close,?= =?UTF-8?q?=20lazy=20import?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - retract_z_if_possible/restore_z_if_captured promoted to module level in objective_turret_controller.py; all four changer classes (NiMotion real+sim, ASI real+sim) delegate — the objective-crash-avoidance policy now has one home instead of four copies. - ASI turret consumes the MS2000Serial transport helpers (wait_idle, command_with_settle_retry, parse_ms2000_number) and utils.resolve_port instead of re-implementing them; unused threading import dropped. - Xeryon changers implement the close() that ObjectiveChangerProtocol always promised (no-op), so the GUI cleanup closes any changer unconditionally — no more vendor-flag OR chain. - ASI turret classes imported inside their construction branch instead of a module-level conditional with None sentinels; the e2e tests drop the class-name monkeypatch boilerplate that pattern forced. - pi.py: dead CombinedStage re-export removed (zero consumers after the composite migration); docstring grammar fixed. - Turret tests use the shared tests.tools fakes (FakeSerialConn, FakeZStage). Co-Authored-By: Claude Fable 5 --- software/control/asi_objective_turret.py | 89 ++++++------------- software/control/gui_hcs.py | 11 ++- software/control/microscope.py | 10 +-- .../objective_changer_2_pos_controller.py | 8 ++ .../control/objective_turret_controller.py | 43 ++++----- software/squid/stage/pi.py | 3 +- .../control/test_asi_objective_turret.py | 44 +-------- .../test_objective_turret_controller.py | 21 +---- software/tests/squid/test_stage.py | 2 +- 9 files changed, 72 insertions(+), 159 deletions(-) diff --git a/software/control/asi_objective_turret.py b/software/control/asi_objective_turret.py index 81ce125f9..0865e24ef 100644 --- a/software/control/asi_objective_turret.py +++ b/software/control/asi_objective_turret.py @@ -12,22 +12,22 @@ startup flow's ``move_to_objective(DEFAULT_OBJECTIVE)`` establishes a known slot at boot. """ -import re -import threading -import time from contextlib import suppress from typing import Dict, Optional import squid.abc import squid.logging -from control.objective_turret_controller import _resolve_position # same KeyError message the GUI dialog shows -from squid.stage.asi import MS2000Serial +from control.objective_turret_controller import ( # shared changer helpers (same KeyError the GUI dialog shows) + _resolve_position, + restore_z_if_captured, + retract_z_if_possible, +) +from squid.stage.asi import MS2000Serial, parse_ms2000_number _log = squid.logging.get_logger(__name__) TURRET_SLOT_COUNT = 6 DEFAULT_MOVE_TIMEOUT_S = 30.0 -_STATUS_POLL_PERIOD_S = 0.05 def _validate_positions(positions: Optional[dict]) -> Dict[str, int]: @@ -73,20 +73,17 @@ def __init__( self._owns_serial = False _log.info("ASI turret reusing the Z stage's MS-2000 connection.") else: - if serial_port: - port = serial_port - elif serial_number: - import squid.stage.utils - - port = squid.stage.utils.resolve_serial_port_by_sn( - serial_number, - missing_hint="Verify the MS-2000 controller is powered and enumerates as a USB serial device.", - ) - else: - raise RuntimeError( + import squid.stage.utils + + port = squid.stage.utils.resolve_port( + serial_port, + serial_number, + missing_hint="Verify the MS-2000 controller is powered and enumerates as a USB serial device.", + unset_message=( "Set ASI_OBJECTIVE_TURRET_SN/ASI_OBJECTIVE_TURRET_SERIAL_PORT (or the ASI_Z_* " "equivalents) to locate the MS-2000 controller." - ) + ), + ) _log.info(f"ASI turret opening its own MS-2000 connection on {port} at {baudrate} baud.") self._serial = MS2000Serial.open(port, baudrate=baudrate) self._owns_serial = True @@ -175,25 +172,17 @@ def _probe_slot(self, require_reply: bool = False) -> Optional[int]: unsupported on this controller build, and we fall back to tracking the last commanded slot. """ - reply = self._serial.command(f"W {self._axis}", check_error=False) + reply = self._serial.command_with_settle_retry(f"W {self._axis}", check_error=False) if require_reply and not reply: - # Our own connection: this probe doubles as the comms sanity check (mirrors - # LS50Controller.initialize -- settle, flush, one retry). - time.sleep(0.2) - with suppress(Exception): - self._serial._serial.reset_input_buffer() - reply = self._serial.command(f"W {self._axis}", check_error=False) - if not reply: - raise RuntimeError( - f"MS-2000 did not answer 'W {self._axis}'. Check the baud rate (ASI RS-232 " - f"DIP default is 9600), the port, and that the controller is powered." - ) + # Our own connection: this probe doubles as the comms sanity check. + raise RuntimeError( + f"MS-2000 did not answer 'W {self._axis}'. Check the baud rate (ASI RS-232 " + f"DIP default is 9600), the port, and that the controller is powered." + ) _log.info(f"ASI turret 'W {self._axis}' reply: {reply!r}") - match = re.search(r"-?\d+", reply.replace(":A", "", 1)) if reply else None - if match: - slot = int(match.group(0)) - if 1 <= slot <= TURRET_SLOT_COUNT: - return slot + value = parse_ms2000_number(reply) + if value is not None and value == int(value) and 1 <= int(value) <= TURRET_SLOT_COUNT: + return int(value) return None def _rotate_to_slot(self, slot: int, timeout_s: float) -> None: @@ -212,25 +201,13 @@ def _rotate_to_slot(self, slot: int, timeout_s: float) -> None: def _wait_idle(self, timeout_s: float) -> None: # '/' is controller-global: a concurrently moving Z also reads busy. Acceptable -- # objective changes and Z scans are not concurrent flows. - deadline = time.monotonic() + timeout_s - while "B" in self._serial.command("/").upper(): - if time.monotonic() > deadline: - raise RuntimeError(f"ASI turret did not reach idle within {timeout_s:.1f}s") - time.sleep(_STATUS_POLL_PERIOD_S) + self._serial.wait_idle(timeout_s, "ASI turret") def _retract_z_if_possible(self) -> Optional[float]: - from control._def import HOMING_ENABLED_Z, OBJECTIVE_RETRACTED_POS_MM - - if self._stage is None or not HOMING_ENABLED_Z: - return None - z_mm = self._stage.get_pos().z_mm - self._stage.move_z_to(OBJECTIVE_RETRACTED_POS_MM) - return z_mm + return retract_z_if_possible(self._stage) def _restore_z_if_captured(self, captured_z: Optional[float]) -> None: - if captured_z is None or self._stage is None: - return - self._stage.move_z_to(captured_z) + restore_z_if_captured(self._stage, captured_z) class ASIObjectiveTurretSimulation: @@ -295,15 +272,7 @@ def _require_open(self) -> None: raise RuntimeError("ASI turret (simulated) is closed") def _retract_z_if_possible(self) -> Optional[float]: - from control._def import HOMING_ENABLED_Z, OBJECTIVE_RETRACTED_POS_MM - - if self._stage is None or not HOMING_ENABLED_Z: - return None - z_mm = self._stage.get_pos().z_mm - self._stage.move_z_to(OBJECTIVE_RETRACTED_POS_MM) - return z_mm + return retract_z_if_possible(self._stage) def _restore_z_if_captured(self, captured_z: Optional[float]) -> None: - if captured_z is None or self._stage is None: - return - self._stage.move_z_to(captured_z) + restore_z_if_captured(self._stage, captured_z) diff --git a/software/control/gui_hcs.py b/software/control/gui_hcs.py index e7c62c224..e1e6adabe 100644 --- a/software/control/gui_hcs.py +++ b/software/control/gui_hcs.py @@ -2914,12 +2914,11 @@ def _cleanup_common(self, for_restart: bool = False): else: raise - # Turret close: always release the serial port so the new process (on restart) - # can acquire it, and so the motor is de-energized on full shutdown. Independent - # of Z-retract success — the close path must run even if Z retract failed. - # ASI turret: close() releases the serial only when the turret opened it; a - # connection shared with the Z stage is closed by self.stage.close() below. - if (USE_OBJECTIVE_TURRET or USE_ASI_OBJECTIVE_TURRET) and self.objective_changer: + # Changer close: every ObjectiveChangerProtocol implementation provides close() + # (release serial ports for a restarted process, de-energize motors; no-op for + # Xeryon). Runs independent of Z-retract success. The ASI turret only closes a + # serial it opened itself; one shared with the Z stage is closed by stage.close(). + if self.objective_changer: try: self.objective_changer.close() except Exception: diff --git a/software/control/microscope.py b/software/control/microscope.py index ec5ff37ef..63aaa6527 100644 --- a/software/control/microscope.py +++ b/software/control/microscope.py @@ -52,12 +52,6 @@ else: ObjectiveTurret4PosController = None -if control._def.USE_ASI_OBJECTIVE_TURRET: - from control.asi_objective_turret import ASIObjectiveTurret, ASIObjectiveTurretSimulation -else: - ASIObjectiveTurret = None - ASIObjectiveTurretSimulation = None - class ObjectiveChangerProtocol(Protocol): """Methods shared by both Xeryon and turret controllers. Controller-specific @@ -182,6 +176,10 @@ def build_from_global_config( else ObjectiveTurret4PosControllerSimulation(**turret_kwargs) ) elif control._def.USE_ASI_OBJECTIVE_TURRET: + # Imported here (not at module level behind the flag) so tests can exercise this + # path by monkeypatching the flag alone. + from control.asi_objective_turret import ASIObjectiveTurret, ASIObjectiveTurretSimulation + if objective_changer_simulated: objective_changer = ASIObjectiveTurretSimulation( positions=control._def.ASI_OBJECTIVE_TURRET_POSITIONS, diff --git a/software/control/objective_changer_2_pos_controller.py b/software/control/objective_changer_2_pos_controller.py index bf2709e07..da4df1c1e 100644 --- a/software/control/objective_changer_2_pos_controller.py +++ b/software/control/objective_changer_2_pos_controller.py @@ -36,6 +36,11 @@ def moveToZero(self): self.axisX.setDPOS(0) self.current_position = 0 + def close(self): + # ObjectiveChangerProtocol promises close(); the Xeryon library holds no OS handle + # that needs explicit release (cleanup zeroes the axis via moveToZero instead). + pass + def moveToPosition1(self, move_z=True): self.axisX.setDPOS(self.position1) if self.stage is not None and self.current_position == 2 and self.retracted: @@ -93,6 +98,9 @@ def __init__(self, sn: str, stage: Optional[squid.abc.AbstractStage] = None): def home(self): pass + def close(self): + pass + def moveToPosition1(self, move_z=True): if self.stage is not None and self.current_position == 2 and self.retracted: # revert retracting z by self.position2_offset diff --git a/software/control/objective_turret_controller.py b/software/control/objective_turret_controller.py index 6f8107f43..1a756b8aa 100644 --- a/software/control/objective_turret_controller.py +++ b/software/control/objective_turret_controller.py @@ -118,6 +118,24 @@ ] +def retract_z_if_possible(stage) -> "Optional[float]": + """Shared objective-changer Z safety: if a stage with Z homing is usable, capture the + current Z and retract to OBJECTIVE_RETRACTED_POS_MM. Returns the captured Z, else None.""" + from control._def import HOMING_ENABLED_Z, OBJECTIVE_RETRACTED_POS_MM + + if stage is None or not HOMING_ENABLED_Z: + return None + z_mm = stage.get_pos().z_mm + stage.move_z_to(OBJECTIVE_RETRACTED_POS_MM) + return z_mm + + +def restore_z_if_captured(stage, captured_z) -> None: + if captured_z is None or stage is None: + return + stage.move_z_to(captured_z) + + def _resolve_position(objective_name: str, positions: dict) -> int: try: return positions[objective_name] @@ -228,19 +246,10 @@ def _require_open(self) -> None: raise RuntimeError("Turret controller is closed") def _retract_z_if_possible(self) -> Optional[float]: - """If stage + Z homing are usable, capture Z and move to safe retract. Return captured z, else None.""" - from control._def import HOMING_ENABLED_Z, OBJECTIVE_RETRACTED_POS_MM - - if self._stage is None or not HOMING_ENABLED_Z: - return None - z_mm = self._stage.get_pos().z_mm - self._stage.move_z_to(OBJECTIVE_RETRACTED_POS_MM) - return z_mm + return retract_z_if_possible(self._stage) def _restore_z_if_captured(self, captured_z: Optional[float]) -> None: - if captured_z is None or self._stage is None: - return - self._stage.move_z_to(captured_z) + restore_z_if_captured(self._stage, captured_z) class ObjectiveTurret4PosController: @@ -415,18 +424,10 @@ def _rotate_to(self, objective_name: str, timeout_s: float) -> None: ) def _retract_z_if_possible(self) -> Optional[float]: - from control._def import HOMING_ENABLED_Z, OBJECTIVE_RETRACTED_POS_MM - - if self._stage is None or not HOMING_ENABLED_Z: - return None - z_mm = self._stage.get_pos().z_mm - self._stage.move_z_to(OBJECTIVE_RETRACTED_POS_MM) - return z_mm + return retract_z_if_possible(self._stage) def _restore_z_if_captured(self, captured_z: Optional[float]) -> None: - if captured_z is None or self._stage is None: - return - self._stage.move_z_to(captured_z) + restore_z_if_captured(self._stage, captured_z) def _calibrate_one( self, diff --git a/software/squid/stage/pi.py b/software/squid/stage/pi.py index 921d899b3..239989d29 100644 --- a/software/squid/stage/pi.py +++ b/software/squid/stage/pi.py @@ -4,7 +4,7 @@ Provides (1) ``C414FocusStage``, a GCS-2.0 driver over a serial (FTDI VCP) link via pipython's pure-Python PISerial gateway (no proprietary PI GCS DLL required), with ``pipython`` imported lazily so this module imports fine without it; (2) ``_SimulatedC414``, -a pure-Python stand-in for hardware-free / CI use; and (3) two ``squid.abc.AbstractStage`` +a pure-Python stand-in for hardware-free / CI use; and (3) the ``squid.abc.AbstractStage`` adapter ``PIFocusStage`` (Z-only; pair with an XY stage via ``squid.stage.composite.CombinedStage``). Z is pure pass-through mm: the controller's native absolute mm is Squid's Z mm, with no sign @@ -21,7 +21,6 @@ import squid.logging from squid.abc import AbstractStage, Pos, StageStage from squid.config import StageConfig -from squid.stage.composite import CombinedStage # re-export: CombinedStage lived here until it went multi-vendor _log = squid.logging.get_logger(__name__) diff --git a/software/tests/control/test_asi_objective_turret.py b/software/tests/control/test_asi_objective_turret.py index aae48a3ee..bed34d737 100644 --- a/software/tests/control/test_asi_objective_turret.py +++ b/software/tests/control/test_asi_objective_turret.py @@ -3,49 +3,11 @@ import control._def import control.asi_objective_turret as aot from squid.stage.asi import MS2000Serial +from tests.tools import FakeSerialConn as _FakeSerialConn, FakeZStage as FakeStage POSITIONS = {"2x": 1, "4x": 2, "10x": 3, "20x": 4, "40x": 5, "60x": 6} -class _FakeSerialConn: - """Scripted pyserial-like object: records writes, pops queued replies, tracks close.""" - - def __init__(self, replies=(), default=b""): - self.written = [] - self.replies = list(replies) - self.default = default - self.closed = False - - def write(self, data): - self.written.append(data) - - def read_until(self, expected=b"\n"): - return self.replies.pop(0) if self.replies else self.default - - def close(self): - self.closed = True - - -class FakeStage: - """Records move_z_to calls and reports a preset Z position.""" - - def __init__(self, z_mm: float = 3.5): - self._z_mm = z_mm - self.z_moves: list[float] = [] - - def move_z_to(self, abs_mm: float, blocking: bool = True): - self.z_moves.append(abs_mm) - self._z_mm = abs_mm - - def get_pos(self): - class _Pos: - pass - - p = _Pos() - p.z_mm = self._z_mm - return p - - def _shared_turret(replies, stage=None, positions=POSITIONS, **kwargs): """Turret on a shared (not owned) scripted transport; first reply feeds the ctor W T probe.""" conn = _FakeSerialConn(replies=replies) @@ -250,7 +212,6 @@ def test_turret_without_port_or_sn_raises(): def _build_asi_scope(monkeypatch, skip_init): import control.microscope - from control.asi_objective_turret import ASIObjectiveTurret, ASIObjectiveTurretSimulation monkeypatch.setattr(control._def, "USE_ASI_Z_STAGE", True, raising=False) monkeypatch.setattr(control._def, "SIMULATE_ASI_Z_STAGE", True, raising=False) @@ -259,9 +220,6 @@ def _build_asi_scope(monkeypatch, skip_init): monkeypatch.setattr(control._def, "ASI_Z_TRAVEL_MM", 50.0, raising=False) monkeypatch.setattr(control._def, "USE_ASI_OBJECTIVE_TURRET", True, raising=False) monkeypatch.setattr(control._def, "SIMULATE_OBJECTIVE_CHANGER", True, raising=False) - # The conditional import left these None (flag was off at import time). - monkeypatch.setattr(control.microscope, "ASIObjectiveTurret", ASIObjectiveTurret, raising=False) - monkeypatch.setattr(control.microscope, "ASIObjectiveTurretSimulation", ASIObjectiveTurretSimulation, raising=False) return control.microscope.Microscope.build_from_global_config(simulated=True, skip_init=skip_init) diff --git a/software/tests/control/test_objective_turret_controller.py b/software/tests/control/test_objective_turret_controller.py index d3cbbf53f..3f7bd05b6 100644 --- a/software/tests/control/test_objective_turret_controller.py +++ b/software/tests/control/test_objective_turret_controller.py @@ -6,6 +6,7 @@ import control._def import control.objective_turret_controller as otc +from tests.tools import FakeZStage as FakeStage from control._def import OBJECTIVE_TURRET_POSITIONS, OBJECTIVE_RETRACTED_POS_MM from control.objective_turret_controller import ( ObjectiveTurret4PosController, @@ -20,26 +21,6 @@ ) -class FakeStage: - """Records move_z_to calls and reports a preset Z position.""" - - def __init__(self, z_mm: float = 3.5): - self._z_mm = z_mm - self.z_moves: list[float] = [] - - def move_z_to(self, abs_mm: float, blocking: bool = True): - self.z_moves.append(abs_mm) - self._z_mm = abs_mm - - def get_pos(self): - class _Pos: - pass - - p = _Pos() - p.z_mm = self._z_mm - return p - - def _make_sim(stage=None): return ObjectiveTurret4PosControllerSimulation( serial_number="SIM-001", diff --git a/software/tests/squid/test_stage.py b/software/tests/squid/test_stage.py index 4296d5073..deda4999d 100644 --- a/software/tests/squid/test_stage.py +++ b/software/tests/squid/test_stage.py @@ -440,7 +440,7 @@ def _sim_asi_stage(**kwargs): def _sim_asi_combined(): - """Simulated Cephla XY + ASI LS50 Z via the reused pi.CombinedStage; returns (combined, xy, z).""" + """Simulated Cephla XY + ASI LS50 Z via squid.stage.composite.CombinedStage; returns (combined, xy, z).""" return _sim_combined_stage(z_stage=_sim_asi_stage(invert_z=True, home_mm=0.0)) From 6bb0fdf84b21ae7bfba65f0b3229dda73ded1c57 Mon Sep 17 00:00:00 2001 From: You Yan Date: Sun, 12 Jul 2026 13:16:21 -0700 Subject: [PATCH 7/9] =?UTF-8?q?fix(turret):=20axis=20letter=20is=20F=20on?= =?UTF-8?q?=20the=20MFC-2000=20=E2=80=94=20default=20M=20F=20/=20W=20F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hardware-confirmed: the objective turret rides the F axis, not T. The configurable ASI_OBJECTIVE_TURRET_AXIS_LETTER now defaults to F, along with the driver/tool defaults, tests, and prose. Co-Authored-By: Claude Fable 5 --- software/control/_def.py | 3 ++- software/control/asi_objective_turret.py | 16 ++++++++-------- .../tests/control/test_asi_objective_turret.py | 16 ++++++++-------- software/tools/asi_z_bringup.py | 2 +- 4 files changed, 19 insertions(+), 18 deletions(-) diff --git a/software/control/_def.py b/software/control/_def.py index f6b456cf0..fa9b1ad01 100644 --- a/software/control/_def.py +++ b/software/control/_def.py @@ -1320,7 +1320,8 @@ def get_wellplate_settings(wellplate_format): # establishes a known slot at every boot. USE_ASI_OBJECTIVE_TURRET = False # Axis letter of the turret on the controller ('M =' / 'W '). -ASI_OBJECTIVE_TURRET_AXIS_LETTER = "T" +# F on the MFC-2000 (hardware-confirmed). +ASI_OBJECTIVE_TURRET_AXIS_LETTER = "F" # Objective name -> turret slot (1..6). Keys must match objectives.csv names. Override per machine .ini # (JSON with double-quoted keys, e.g. asi_objective_turret_positions = {"4x": 1, "20x": 2}). ASI_OBJECTIVE_TURRET_POSITIONS = {"2x": 1, "4x": 2, "10x": 3, "20x": 4, "40x": 5, "60x": 6} diff --git a/software/control/asi_objective_turret.py b/software/control/asi_objective_turret.py index 0865e24ef..64a53e0e0 100644 --- a/software/control/asi_objective_turret.py +++ b/software/control/asi_objective_turret.py @@ -1,8 +1,8 @@ """6-position ASI objective turret on an MS-2000-family controller. Shares the controller -- and, when USE_ASI_Z_STAGE is enabled, the serial connection -- with -the LS50 Z stage (squid.stage.asi). Command set: ``M T=`` rotates to a RAW slot index -1..6 (NOT scaled by the 0.1 um unit factor used for linear axes); ``W T`` presumably reads +the LS50 Z stage (squid.stage.asi). Command set: ``M F=`` rotates to a RAW slot index +1..6 (NOT scaled by the 0.1 um unit factor used for linear axes); ``W F`` presumably reads the slot back (UNVERIFIED on hardware -- treated as best-effort, degrading to software tracking of the last commanded slot); ``/`` is the controller-GLOBAL busy byte, shared with the Z axis. ``MS2000Serial``'s internal lock keeps turret and Z commands frame-safe when @@ -35,7 +35,7 @@ def _validate_positions(positions: Optional[dict]) -> Dict[str, int]: from control._def import ASI_OBJECTIVE_TURRET_POSITIONS positions = ASI_OBJECTIVE_TURRET_POSITIONS - # Fail fast: a bad slot value would otherwise go straight to the hardware as 'M T='. + # Fail fast: a bad slot value would otherwise go straight to the hardware as 'M F='. for name, slot in positions.items(): if not isinstance(slot, int) or not 1 <= slot <= TURRET_SLOT_COUNT: raise ValueError( @@ -45,7 +45,7 @@ def _validate_positions(positions: Optional[dict]) -> Dict[str, int]: class ASIObjectiveTurret: - """ObjectiveChangerProtocol implementation for the MS-2000 turret (T) axis. + """ObjectiveChangerProtocol implementation for the MS-2000 turret axis (F on the MFC-2000). Pass ``shared_serial`` (the Z stage's transport, via squid.stage.asi.find_shared_ms2000) when the LS50 Z stage runs on the same controller -- the turret then never closes it. @@ -58,7 +58,7 @@ def __init__( serial_number: Optional[str] = None, serial_port: Optional[str] = None, baudrate: int = 115200, - axis: str = "T", + axis: str = "F", positions: Optional[Dict[str, int]] = None, stage: Optional[squid.abc.AbstractStage] = None, ): @@ -166,9 +166,9 @@ def _require_open(self) -> None: raise RuntimeError("ASI turret is closed") def _probe_slot(self, require_reply: bool = False) -> Optional[int]: - """Best-effort 'W T' read. Returns the slot only for a plausible 1..N integer reply. + """Best-effort 'W F' read. Returns the slot only for a plausible 1..N integer reply. - check_error=False: an ':N-...' ack still proves comms -- it just means 'W T' is + check_error=False: an ':N-...' ack still proves comms -- it just means 'W F' is unsupported on this controller build, and we fall back to tracking the last commanded slot. """ @@ -217,7 +217,7 @@ def __init__( self, positions: Optional[Dict[str, int]] = None, stage: Optional[squid.abc.AbstractStage] = None, - axis: str = "T", + axis: str = "F", ): self._positions = _validate_positions(positions) self._axis = axis diff --git a/software/tests/control/test_asi_objective_turret.py b/software/tests/control/test_asi_objective_turret.py index bed34d737..a6fdec8c8 100644 --- a/software/tests/control/test_asi_objective_turret.py +++ b/software/tests/control/test_asi_objective_turret.py @@ -9,7 +9,7 @@ def _shared_turret(replies, stage=None, positions=POSITIONS, **kwargs): - """Turret on a shared (not owned) scripted transport; first reply feeds the ctor W T probe.""" + """Turret on a shared (not owned) scripted transport; first reply feeds the ctor W F probe.""" conn = _FakeSerialConn(replies=replies) turret = aot.ASIObjectiveTurret(shared_serial=MS2000Serial(conn), positions=positions, stage=stage, **kwargs) return turret, conn @@ -119,11 +119,11 @@ def test_invalid_slot_positions_raise(): def test_move_frames_raw_slot_command(): - # Slot index is sent RAW ('M T=3'), not scaled by the 0.1 um unit factor. + # Slot index is sent RAW ('M F=3'), not scaled by the 0.1 um unit factor. turret, conn = _shared_turret([b":A 1\r\n", b":A\r\n", b"N\r\n", b":A 3\r\n"]) - assert turret.current_slot == 1 # seeded from the ctor W T probe + assert turret.current_slot == 1 # seeded from the ctor W F probe turret.move_to_objective("10x") - assert conn.written == [b"W T\r", b"M T=3\r", b"/\r", b"W T\r"] + assert conn.written == [b"W F\r", b"M F=3\r", b"/\r", b"W F\r"] assert turret.current_objective == "10x" assert turret.current_slot == 3 @@ -145,10 +145,10 @@ def test_move_timeout_raises_and_still_restores_z(monkeypatch): def test_probe_seed_short_circuits_move(): - # Ctor W T said slot 4; moving to the slot-4 objective must not rotate. + # Ctor W F said slot 4; moving to the slot-4 objective must not rotate. turret, conn = _shared_turret([b":A 4\r\n"]) turret.move_to_objective("20x") - assert conn.written == [b"W T\r"] # no M command + assert conn.written == [b"W F\r"] # no M command assert turret.current_objective == "20x" @@ -156,7 +156,7 @@ def test_probe_garbage_means_unknown_slot(): turret, conn = _shared_turret([b":N-1\r\n", b":A\r\n", b"N\r\n"]) assert turret.current_slot is None turret.move_to_objective("2x") # unknown never short-circuits: rotation commanded - assert b"M T=1\r" in conn.written + assert b"M F=1\r" in conn.written def test_unknown_objective_keyerror_real(): @@ -192,7 +192,7 @@ def test_owned_ctor_probe_failure_closes_serial(monkeypatch): def test_home_is_motionless_real(): turret, conn = _shared_turret([b":A 2\r\n", b":A 2\r\n"]) turret.home() - assert all(not w.startswith(b"M ") for w in conn.written) # W T probes only, no motion + assert all(not w.startswith(b"M ") for w in conn.written) # W F probes only, no motion assert turret.current_slot == 2 diff --git a/software/tools/asi_z_bringup.py b/software/tools/asi_z_bringup.py index bfd5b73b1..dc30c83be 100644 --- a/software/tools/asi_z_bringup.py +++ b/software/tools/asi_z_bringup.py @@ -193,7 +193,7 @@ def main(): parser.add_argument("--fence-mm", type=float, default=0.0, help="Also test the +/- soft-limit fence (0 = skip)") parser.add_argument("--scan-bauds", action="store_true", help="Try each ASI baud rate and report which answers") parser.add_argument("--turret", action="store_true", help="Probe the objective turret axis (read-only)") - parser.add_argument("--turret-axis", default="T", help="Turret axis letter (default T)") + parser.add_argument("--turret-axis", default="F", help="Turret axis letter (default F, per the MFC-2000)") parser.add_argument("--turret-slot", type=int, help="Rotate to this slot (1..6); requires --allow-motion") parser.add_argument( From 997911878c40693ec1386dd9865ca03520da9e95 Mon Sep 17 00:00:00 2001 From: You Yan Date: Sun, 12 Jul 2026 13:32:45 -0700 Subject: [PATCH 8/9] docs(turret): last M T reference in the _def flag comment -> M F Co-Authored-By: Claude Fable 5 --- software/control/_def.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/software/control/_def.py b/software/control/_def.py index fa9b1ad01..5831ac11c 100644 --- a/software/control/_def.py +++ b/software/control/_def.py @@ -1309,7 +1309,7 @@ def get_wellplate_settings(wellplate_format): # Objective name -> turret slot index (1..4). Override per machine in .ini. OBJECTIVE_TURRET_POSITIONS = {"4x": 1, "10x": 2, "20x": 3, "40x": 4} -# 6-position ASI objective turret on an MS-2000-family controller ('M T=' with a RAW +# 6-position ASI objective turret on an MS-2000-family controller ('M F=' with a RAW # slot index 1..6 -- not scaled like linear-axis units). Typically the SAME controller as the # ASI LS50 Z stage (USE_ASI_Z_STAGE): when both are enabled the turret reuses the Z stage's # serial connection and the SN/port/baud below are ignored. They apply only when the turret From c194115c3016ea06e47468492d97df50e1bb1162 Mon Sep 17 00:00:00 2001 From: You Yan Date: Sun, 19 Jul 2026 20:30:31 -0700 Subject: [PATCH 9/9] fix: address Copilot review comments - ObjectiveChanger2PosController.close() now calls Xeryon.stop() so the serial port and reader thread are released before the restarted process re-acquires the device (the old no-op comment was wrong: Communication holds an open serial.Serial plus a reader thread) - asi_z_bringup: constrain --turret-slot to 1..6 so invalid slots fail in argparse instead of reaching the controller as 'M F=' - composite stage: remove unused _log / squid.logging import Co-Authored-By: Claude Fable 5 --- software/control/gui_hcs.py | 6 ++--- .../objective_changer_2_pos_controller.py | 13 ++++++++--- software/squid/stage/composite.py | 3 --- ...test_objective_changer_2_pos_controller.py | 22 +++++++++++++++++++ software/tools/asi_z_bringup.py | 4 +++- 5 files changed, 38 insertions(+), 10 deletions(-) diff --git a/software/control/gui_hcs.py b/software/control/gui_hcs.py index e1e6adabe..a45d6b54b 100644 --- a/software/control/gui_hcs.py +++ b/software/control/gui_hcs.py @@ -2915,9 +2915,9 @@ def _cleanup_common(self, for_restart: bool = False): raise # Changer close: every ObjectiveChangerProtocol implementation provides close() - # (release serial ports for a restarted process, de-energize motors; no-op for - # Xeryon). Runs independent of Z-retract success. The ASI turret only closes a - # serial it opened itself; one shared with the Z stage is closed by stage.close(). + # (release serial ports for a restarted process, de-energize motors). Runs + # independent of Z-retract success. The ASI turret only closes a serial it + # opened itself; one shared with the Z stage is closed by stage.close(). if self.objective_changer: try: self.objective_changer.close() diff --git a/software/control/objective_changer_2_pos_controller.py b/software/control/objective_changer_2_pos_controller.py index da4df1c1e..c65352a85 100644 --- a/software/control/objective_changer_2_pos_controller.py +++ b/software/control/objective_changer_2_pos_controller.py @@ -6,8 +6,11 @@ from control.Xeryon import Xeryon, Stage from control._def import * # to remove once we create ObjectiveChangerConfig import squid.abc +import squid.logging from typing import Optional +_log = squid.logging.get_logger(__name__) + class ObjectiveChanger2PosController: def __init__(self, sn: str, stage: Optional[squid.abc.AbstractStage] = None): @@ -37,9 +40,13 @@ def moveToZero(self): self.current_position = 0 def close(self): - # ObjectiveChangerProtocol promises close(); the Xeryon library holds no OS handle - # that needs explicit release (cleanup zeroes the axis via moveToZero instead). - pass + # Xeryon.stop() sends STOP to the axis and shuts down the reader thread, which + # closes the serial port — required so a restarted process (spawned while this + # one is still alive) can re-acquire the device. Best-effort during teardown. + try: + self.controller.stop() + except Exception: + _log.exception("Error stopping the Xeryon controller during close()") def moveToPosition1(self, move_z=True): self.axisX.setDPOS(self.position1) diff --git a/software/squid/stage/composite.py b/software/squid/stage/composite.py index abe06aeaf..e9636f5f2 100644 --- a/software/squid/stage/composite.py +++ b/software/squid/stage/composite.py @@ -10,12 +10,9 @@ from typing import Optional -import squid.logging from squid.abc import AbstractStage, Pos, StageStage from squid.config import StageConfig -_log = squid.logging.get_logger(__name__) - class CombinedStage(AbstractStage): """AbstractStage routing X / Y / theta to xy_stage and Z to z_stage (an external Z-only focus stage, e.g. the V-308 or the ASI LS50).""" diff --git a/software/tests/control/test_objective_changer_2_pos_controller.py b/software/tests/control/test_objective_changer_2_pos_controller.py index bc2e73106..ff24a7cfd 100644 --- a/software/tests/control/test_objective_changer_2_pos_controller.py +++ b/software/tests/control/test_objective_changer_2_pos_controller.py @@ -2,8 +2,11 @@ from __future__ import annotations +from unittest.mock import Mock + import control._def from control.objective_changer_2_pos_controller import ( + ObjectiveChanger2PosController, ObjectiveChanger2PosController_Simulation, ) @@ -56,3 +59,22 @@ def test_move_to_objective_unknown_raises(monkeypatch): sim = ObjectiveChanger2PosController_Simulation(sn="SIM", stage=FakeStage()) with pytest.raises(KeyError): sim.move_to_objective("999x") + + +def _controller_with_fake_xeryon() -> ObjectiveChanger2PosController: + # __init__ opens a serial port, so bypass it and inject only what close() touches. + changer = ObjectiveChanger2PosController.__new__(ObjectiveChanger2PosController) + changer.controller = Mock() + return changer + + +def test_close_stops_the_xeryon_controller(): + changer = _controller_with_fake_xeryon() + changer.close() + changer.controller.stop.assert_called_once() + + +def test_close_swallows_stop_errors(): + changer = _controller_with_fake_xeryon() + changer.controller.stop.side_effect = RuntimeError("port already gone") + changer.close() # must not raise during teardown diff --git a/software/tools/asi_z_bringup.py b/software/tools/asi_z_bringup.py index dc30c83be..4d845704d 100644 --- a/software/tools/asi_z_bringup.py +++ b/software/tools/asi_z_bringup.py @@ -194,7 +194,9 @@ def main(): parser.add_argument("--scan-bauds", action="store_true", help="Try each ASI baud rate and report which answers") parser.add_argument("--turret", action="store_true", help="Probe the objective turret axis (read-only)") parser.add_argument("--turret-axis", default="F", help="Turret axis letter (default F, per the MFC-2000)") - parser.add_argument("--turret-slot", type=int, help="Rotate to this slot (1..6); requires --allow-motion") + parser.add_argument( + "--turret-slot", type=int, choices=range(1, 7), help="Rotate to this slot (1..6); requires --allow-motion" + ) parser.add_argument( "--find-zero",