diff --git a/apps/autocal/config/locales/en_US.json b/apps/autocal/config/locales/en_US.json index aebcfb5..77bd784 100644 --- a/apps/autocal/config/locales/en_US.json +++ b/apps/autocal/config/locales/en_US.json @@ -48,13 +48,13 @@ }, "move": { "execute": "Execute", - "execute_hint": "Send the selected point's solved pose to the platform: yaw, then pitch, then table X/Y. The press axis Z is not moved.", + "execute_hint": "Send the selected point's solved pose to the platform: yaw, pitch and table X/Y move together. The press axis Z is not moved.", "stop": "Stop", "stop_hint": "Stop every axis now (the same command as 全部停止 in the strip).", "return_initial": "Return to zero", "return_hint": "Every axis back to the initial state, press axis Z first.", "confirm_title": "Confirm motion", - "confirm_execute": "Will move, in this order:\n\n{steps}\n\nCheck this against the machine before continuing.", + "confirm_execute": "Will move these axes together:\n\n{steps}\n\nCheck this against the machine before continuing.", "confirm_return": "Will move every axis back to zero (press axis Z first):\n\n{steps}", "hold_note": "Not moved: {roles} (the press axis is P7's business — contact detection and press depth)", "cancelled": "Cancelled; nothing was commanded", diff --git a/apps/autocal/config/locales/zh_CN.json b/apps/autocal/config/locales/zh_CN.json index edaf80a..0612ff6 100644 --- a/apps/autocal/config/locales/zh_CN.json +++ b/apps/autocal/config/locales/zh_CN.json @@ -48,13 +48,13 @@ }, "move": { "execute": "执行动作", - "execute_hint": "把选中点解算出的位姿送到平台：先回转、再俯仰，最后工作台 X/Y。压轴 Z 不动作。", + "execute_hint": "把选中点解算出的位姿送到平台：回转、俯仰、工作台 X/Y 同时移动。压轴 Z 不动作。", "stop": "停止", "stop_hint": "立即停止所有轴（与顶栏「全部停止」相同）。", "return_initial": "回初始位", "return_hint": "五个轴全部回到零点，先退压轴 Z。", "confirm_title": "确认运动", - "confirm_execute": "将按以下顺序移动：\n\n{steps}\n\n请确认仿真与机床一致。", + "confirm_execute": "将同时移动：\n\n{steps}\n\n请确认仿真与机床一致。", "confirm_return": "将把五个轴全部移回零点（先退压轴 Z）：\n\n{steps}", "hold_note": "不动作：{roles}（压轴由 P7 的接触检测与压深另外管）", "cancelled": "已取消，未发出任何运动", diff --git a/apps/autocal/src/orisys_autocal/motion/pose_move.py b/apps/autocal/src/orisys_autocal/motion/pose_move.py index 88ad6dd..210db8d 100644 --- a/apps/autocal/src/orisys_autocal/motion/pose_move.py +++ b/apps/autocal/src/orisys_autocal/motion/pose_move.py @@ -1,4 +1,4 @@ -"""Moving the platform to a simulated pose, one axis at a time. +"""Moving the platform to a simulated pose. The finger screen solves a pose — rotate the stages, slide the table — and shows it. Executing it is a *sequence* of coordinated moves, and where that sequence lives @@ -10,9 +10,10 @@ matters: :class:`~orisys_autocal.motion.poller.MotionPoller` is built, and it is why a stop is possible at all: a mover that sat in a loop inside one command could not be interrupted, because the stop would have to wait for the loop it is meant to abort. -* **This module** owns the order and the waiting. It watches the snapshots the poller - already produces, issues the next axis when the current one has arrived, and gives - up on its own if a step stops making progress. +* **This module** owns the waiting. An execute plan starts yaw, pitch and table X/Y + in one command so they move together; a return plan still retracts one axis at a + time. It watches the snapshots the poller already produces and gives up on its + own if an axis stops short of its target. That shape is the whole reason for the class: ``stop`` is submitted as an ordinary command, so it runs at the next tick and stops the axis that is moving *now*. @@ -27,7 +28,7 @@ from __future__ import annotations import time from collections.abc import Mapping, Sequence -from dataclasses import dataclass +from dataclasses import dataclass, field from PySide6.QtCore import QObject, Signal @@ -80,6 +81,9 @@ class MovePlan: #: Roles deliberately left alone. Reported, because "z did not move" is #: information the operator needs, not an omission. held: tuple[str, ...] = () + #: When set, every step is commanded in one call and they travel together. + #: Return-to-zero leaves this off so the press axis clears before anything turns. + together: bool = False def describe(self) -> str: parts = [step.label() for step in self.steps] @@ -100,6 +104,7 @@ class MoveState: failure: str = "" cancelled: bool = False _idle_since: float | None = None + _group_idle: dict = field(default_factory=dict) @property def step(self) -> MoveStep | None: @@ -119,6 +124,7 @@ def plan_pose_move( title: str = "", hold: Sequence[str] = (), order: Sequence[str] = MOVE_ORDER, + together: bool = False, ) -> MovePlan: """Turn per-role targets into a stepped plan, in ``order``. @@ -143,7 +149,7 @@ def plan_pose_move( unit="mm" if role in LINEAR_ROLES else "°", ) ) - return MovePlan(steps=tuple(steps), title=title, held=tuple(hold)) + return MovePlan(steps=tuple(steps), title=title, held=tuple(hold), together=together) def plan_return_to_initial(service: MotionService) -> MovePlan: @@ -220,6 +226,10 @@ class PoseMover(QObject): self.cancelled.emit() def _issue_current(self) -> None: + plan = self.state.plan + if plan is not None and plan.together: + self._issue_together(plan) + return step = self.state.step if step is None: self._complete() @@ -231,6 +241,19 @@ class PoseMover(QObject): self._submit(lambda service, step=step: service.move_abs(step.axis, step.units)) self.stepStarted.emit(step) + def _issue_together(self, plan: MovePlan) -> None: + """Command every step in one worker call. Z is absent from the plan.""" + moves: list[tuple[int, float]] = [] + for step in plan.steps: + if not step.units == step.units: + self._fail(f"{step.role} 无法换算成用户单位 / cannot convert {step.role}") + return + moves.append((step.axis, step.units)) + self.state._group_idle = {} + self._submit(lambda service, moves=tuple(moves): service.move_abs_many(moves)) + if plan.steps: + self.stepStarted.emit(plan.steps[0]) + def _complete(self) -> None: self.state.finished = True plan = self.state.plan @@ -248,6 +271,10 @@ class PoseMover(QObject): """Advance the plan from a poll. Called on every refresh.""" if not self.running: return + plan = self.state.plan + if plan is not None and plan.together: + self._on_snapshot_together(snapshot) + return step = self.state.step if step is None: self._complete() @@ -289,12 +316,58 @@ class PoseMover(QObject): f"{step.target:.4f}{step.unit} / stopped short of target" ) + def _on_snapshot_together(self, snapshot: Snapshot) -> None: + """Finish when every commanded axis has arrived. Z is not in the plan.""" + plan = self.state.plan + if plan is None: + return + now = time.monotonic() + pending = False + for step in plan.steps: + state = snapshot.axis(step.axis) + if state is None: + return + if state.error or not state.read_ok: + self._fail(f"{step.role}: {state.error or 'read failed'}") + return + if state.status.alarm: + self._fail(f"{step.role} 报警 / alarm on {step.role}") + return + if step.unit == "mm": + travelled = self._service.to_mm(step.axis, state.dpos) + else: + travelled = user_units_to_deg(step.axis, state.dpos) + tolerance = SETTLE_MM if step.unit == "mm" else SETTLE_DEG + arrived = state.idle and abs(travelled - step.target) <= tolerance + if arrived: + self.state._group_idle.pop(step.axis, None) + continue + pending = True + if not state.idle: + self.state._group_idle.pop(step.axis, None) + continue + since = self.state._group_idle.get(step.axis) + if since is None: + self.state._group_idle[step.axis] = now + elif now - since > STALL_GRACE_S: + self._fail( + f"{step.role} 停在 {travelled:.4f}{step.unit}，目标 " + f"{step.target:.4f}{step.unit} / stopped short of target" + ) + return + if not pending: + self._complete() + def progress_text(self) -> str: """A one-line status for the panel.""" if self.state.failure: return f"✗ {self.state.failure}" if self.state.finished: return "✓ 已完成 / done" + plan = self.state.plan + if plan is not None and plan.together: + roles = " ".join(step.role for step in plan.steps) + return f"同时 {roles}" step = self.state.step if step is None: return "" diff --git a/apps/autocal/src/orisys_autocal/motion/service.py b/apps/autocal/src/orisys_autocal/motion/service.py index 1920214..ecb7278 100644 --- a/apps/autocal/src/orisys_autocal/motion/service.py +++ b/apps/autocal/src/orisys_autocal/motion/service.py @@ -17,7 +17,7 @@ import logging import math import os import time -from collections.abc import Callable, Iterable, Mapping +from collections.abc import Callable, Iterable, Mapping, Sequence from orisys_app_shared.zmotion import ZAuxError @@ -399,12 +399,20 @@ class MotionService: return MotionResult.failed(code, message) self._spec = spec - logger.info("Connected: %s via %s", spec.describe(), self._transport.source) # A controller may already have a BAS program running from a previous # session; adopt it rather than claiming the bus is uninitialised. The # adopted state is *not* marked verified: its globals are a latch, and # nothing in this session has yet seen a scan happen. + # The remembered file and the preference-file axis parameters stay on + # the PC. Writing either here would replace configuration set from the + # vendor software, so connect only reads. self._adopt_existing_bas(bas_path=bas_path) + logger.info( + "Connected: %s via %s. BAS was not downloaded and axis parameters were not written.", + spec.describe(), + self._transport.source, + ) + self._log_axis_readback("connect") return MotionResult.success(f"已连接 {spec.describe()}" if self._is_zh else f"Connected {spec.describe()}") @property @@ -748,7 +756,17 @@ class MotionService: code = getattr(exc, "code", 0) return MotionResult.failed(code, self._msg(code) if code else str(exc)) self._params[axis] = params - return MotionResult.success() + result = MotionResult.success() + logger.info( + "Wrote units/speed/accel on axis %s (units=%s speed=%s accel=%s); " + "drive profile and following-error limit were not written. %s", + axis, + params.units, + params.speed, + params.accel, + self._position_text(axis), + ) + return result def set_axis_enable(self, axis: int, enabled: bool) -> MotionResult: if not self.connected: @@ -759,10 +777,18 @@ class MotionService: self._transport.set_axis_enable(axis, enabled) except (ZAuxError, OSError) as exc: code = getattr(exc, "code", 0) - return MotionResult.failed(code, self._msg(code) if code else str(exc)) - return MotionResult.success( - self._text(f"轴{axis} 已{'使能' if enabled else '关闭'}", - f"Axis {axis} {'enabled' if enabled else 'disabled'}") + return self._logged( + "enable", + axis, + MotionResult.failed(code, self._msg(code) if code else str(exc)), + ) + return self._logged( + "enable", + axis, + MotionResult.success( + self._text(f"轴{axis} 已{'使能' if enabled else '关闭'}", + f"Axis {axis} {'enabled' if enabled else 'disabled'}") + ), ) def clear_alarm(self, axis: int) -> MotionResult: @@ -774,8 +800,16 @@ class MotionService: self._transport.bus_drive_clear(axis, 0) except (ZAuxError, OSError) as exc: code = getattr(exc, "code", 0) - return MotionResult.failed(code, self._msg(code) if code else str(exc)) - return MotionResult.success(self._text("已清除报警", "Alarm cleared")) + return self._logged( + "clear_alarm", + axis, + MotionResult.failed(code, self._msg(code) if code else str(exc)), + ) + return self._logged( + "clear_alarm", + axis, + MotionResult.success(self._text("已清除报警", "Alarm cleared")), + ) # ------------------------------------------------------------------ motion @@ -797,16 +831,60 @@ class MotionService: def jog(self, axis: int, forward: bool) -> MotionResult: refusal = self._motion_guard(axis) if refusal is not None: - return refusal + return self._logged("jog", axis, refusal) try: self._transport.jog(axis, forward) except (ZAuxError, OSError) as exc: code = getattr(exc, "code", 0) - return MotionResult.failed(code, self._msg(code) if code else str(exc)) - return MotionResult.success( - self._text("正转" if forward else "反转", "Forward" if forward else "Reverse") + return self._logged( + "jog", + axis, + MotionResult.failed(code, self._msg(code) if code else str(exc)), + ) + return self._logged( + "jog", + axis, + MotionResult.success( + self._text("正转" if forward else "反转", "Forward" if forward else "Reverse") + ), ) + def move_abs_many(self, moves: Sequence[tuple[int, float]]) -> MotionResult: + """Absolute-move several axes in one call so they travel together. + + Guards are checked for every axis before the first command is sent, so a + refusal does not leave the earlier axes already moving. Each move still + uses the same absolute-move call as a single axis. + """ + pending = [(int(axis), float(value)) for axis, value in moves] + if not pending: + return MotionResult.refused(self._text("没有要移动的轴", "Nothing to move")) + for axis, value in pending: + refusal = self._motion_guard(axis) + if refusal is not None: + return self._logged("move_abs", axis, refusal) + if not math.isfinite(value): + return self._logged( + "move_abs", + axis, + MotionResult.refused( + self._text("目标位置请输入数字", "Enter a numeric position") + ), + ) + last = MotionResult.success() + for axis, value in pending: + try: + self._transport.move_abs(axis, value) + except (ZAuxError, OSError) as exc: + code = getattr(exc, "code", 0) + return self._logged( + "move_abs", + axis, + MotionResult.failed(code, self._msg(code) if code else str(exc)), + ) + last = self._logged("move_abs", axis, MotionResult.success()) + return last + def move_abs(self, axis: int, target: float) -> MotionResult: return self._move(axis, target, absolute=True) @@ -814,12 +892,17 @@ class MotionService: return self._move(axis, distance, absolute=False) def _move(self, axis: int, value: float, *, absolute: bool) -> MotionResult: + action = "move_abs" if absolute else "move_rel" refusal = self._motion_guard(axis) if refusal is not None: - return refusal + return self._logged(action, axis, refusal) if not isinstance(value, (int, float)) or not math.isfinite(float(value)): - return MotionResult.refused( - self._text("目标位置请输入数字", "Enter a numeric position") + return self._logged( + action, + axis, + MotionResult.refused( + self._text("目标位置请输入数字", "Enter a numeric position") + ), ) try: if absolute: @@ -828,8 +911,12 @@ class MotionService: self._transport.move_rel(axis, float(value)) except (ZAuxError, OSError) as exc: code = getattr(exc, "code", 0) - return MotionResult.failed(code, self._msg(code) if code else str(exc)) - return MotionResult.success() + return self._logged( + action, + axis, + MotionResult.failed(code, self._msg(code) if code else str(exc)), + ) + return self._logged(action, axis, MotionResult.success()) def stop(self, axis: int) -> MotionResult: """Stop one axis (decelerating stop, mode 2).""" @@ -841,8 +928,16 @@ class MotionService: self._transport.stop(axis, 2) except (ZAuxError, OSError) as exc: code = getattr(exc, "code", 0) - return MotionResult.failed(code, self._msg(code) if code else str(exc)) - return MotionResult.success(self._text("已停止", "Stopped")) + return self._logged( + "stop", + axis, + MotionResult.failed(code, self._msg(code) if code else str(exc)), + ) + return self._logged( + "stop", + axis, + MotionResult.success(self._text("已停止", "Stopped")), + ) def stop_all(self) -> list[MotionResult]: """Stop every axis. Used before disconnecting.""" @@ -882,6 +977,46 @@ class MotionService: self._home_done[axis] = False return MotionResult.success(self._text("回零已启动", "Homing started")) + def _log_axis_readback(self, action: str) -> None: + """Read every axis and log it. Does not write parameters or start motion.""" + for axis in range(self.axis_count): + self._log_command(action, axis, MotionResult.success("read")) + + def _logged(self, action: str, axis: int, result: MotionResult) -> MotionResult: + self._log_command(action, axis, result) + return result + + def _log_command(self, action: str, axis: int, result: MotionResult) -> None: + text = ( + f"{action} axis {axis} -> ok={result.ok} code={result.code} " + f"message={result.message!r} {self._position_text(axis)}" + ) + if result.ok: + logger.info(text) + else: + logger.warning(text) + + def _position_text(self, axis: int) -> str: + """Commanded and measured position, for the command log. + + Reads only. A failed read is part of the line rather than a second error, + because the command result itself is what the caller is waiting on. + """ + if not self.connected: + return "dpos=unread mpos=unread status=unread enable=unread" + try: + return ( + f"dpos={self._transport.get_dpos(axis)} " + f"mpos={self._transport.get_mpos(axis)} " + f"status={self._transport.get_axis_status(axis)} " + f"enable={self._transport.get_axis_enable(axis)} " + f"atype={self._transport.get_atype(axis)} " + f"units={self._transport.get_units(axis)} " + f"speed={self._transport.get_speed(axis)}" + ) + except (ZAuxError, OSError) as exc: + return f"dpos=unread mpos=unread ({exc})" + def _known_axis(self, axis: int) -> bool: return 0 <= int(axis) < self.axis_count diff --git a/apps/autocal/src/orisys_autocal/ui/axis_panel.py b/apps/autocal/src/orisys_autocal/ui/axis_panel.py index b6b0b85..dfbd5ff 100644 --- a/apps/autocal/src/orisys_autocal/ui/axis_panel.py +++ b/apps/autocal/src/orisys_autocal/ui/axis_panel.py @@ -372,7 +372,20 @@ class AxisPanel(QGroupBox): self._recolour_rows() def _axis_present(self, row: int) -> bool: - return not self._available_axes or row < self._available_axes + """Rows the bus counted, plus any axis the controller has already mapped. + + ``Bus_TotalAxisnum`` is a global left by the last scan. It can stay + short of the real axis list after a drive is brought back without + another scan, which disabled axis 4 while that same controller axis + still answered and could be moved from the vendor software. A non-zero + axis type is that answer. Axes 0-3 already pass the count check; axis + 4 passes this one once it is mapped. An unmapped axis (type 0) stays + disabled, and none of this writes parameters or starts motion. + """ + if not self._available_axes or row < self._available_axes: + return True + state = self._snapshot.axis(row) + return state is not None and state.read_ok and state.atype != 0 def _refresh_availability(self) -> None: """Recompute every control's enabled state from connection + bus.""" diff --git a/apps/autocal/src/orisys_autocal/ui/window.py b/apps/autocal/src/orisys_autocal/ui/window.py index a18d269..e04efff 100644 --- a/apps/autocal/src/orisys_autocal/ui/window.py +++ b/apps/autocal/src/orisys_autocal/ui/window.py @@ -1013,10 +1013,12 @@ class AutoCalMainWindow(QMainWindow): self._pending[token] = "" def _on_jog(self, axis: int, forward: bool) -> None: - self.poller.submit(lambda service: service.jog(axis, forward)) + token = self.poller.submit(lambda service: service.jog(axis, forward)) + self._pending[token] = "" def _on_stop(self, axis: int) -> None: - self.poller.submit(lambda service: service.stop(axis)) + token = self.poller.submit(lambda service: service.stop(axis)) + self._pending[token] = "" def _on_stop_all(self) -> None: def _stop_all(service: MotionService) -> object: @@ -1043,10 +1045,10 @@ class AutoCalMainWindow(QMainWindow): def _on_execute_pose(self) -> None: """Move the platform to the solved pose of the finger screen's selection. - Stages and table only: the press axis is deliberately not commanded, so this - is workflow P6's dry run for one point — it cannot reach the sensor. The - numbers shown in the confirmation are the ones the preview drew, so the - operator checks the simulation against the machine, not a summary of it. + Yaw, pitch and table X/Y start together. The press axis is deliberately + not commanded, so this is workflow P6's dry run for one point — it cannot + reach the sensor. The numbers shown in the confirmation are the ones the + preview drew, so the operator checks the simulation against the machine. """ solve = self.finger_panel.solved_pose() if solve is None: @@ -1058,6 +1060,7 @@ class AutoCalMainWindow(QMainWindow): targets, title=t("move.execute"), hold=held, + together=True, ) self._start_move(plan, t("move.execute")) @@ -1080,8 +1083,15 @@ class AutoCalMainWindow(QMainWindow): # cannot be turned off is one that gets clicked through unread. if self._confirm_moves: steps = "\n".join( - f" {index + 1}. {step.role} → {step.target:+.4f} {step.unit}" - f" (驱动器 {step.axis})" + ( + f" {step.role} → {step.target:+.4f} {step.unit}" + f" (驱动器 {step.axis})" + if plan.together + else ( + f" {index + 1}. {step.role} → {step.target:+.4f} {step.unit}" + f" (驱动器 {step.axis})" + ) + ) for index, step in enumerate(plan.steps) ) if plan.held: diff --git a/apps/autocal/tests/test_controller_sequence.py b/apps/autocal/tests/test_controller_sequence.py index 7f41007..f2c6682 100644 --- a/apps/autocal/tests/test_controller_sequence.py +++ b/apps/autocal/tests/test_controller_sequence.py @@ -17,6 +17,7 @@ No test here touches a controller or a DLL. from __future__ import annotations import ctypes +import logging import time from pathlib import Path @@ -25,6 +26,7 @@ from orisys_autocal.motion import ( DEFAULT_BAS_TASK, INIT_OK, INIT_RUNNING, + BusStatus, ConnectionSpec, DemoTransport, MotionService, @@ -224,6 +226,44 @@ def test_adopted_program_without_a_known_file_says_so(): assert "文件名" in result.message or "file name" in result.message +def test_connect_reads_axis_4_and_does_not_write_parameters(caplog): + """Connecting must not replace configuration the vendor software just set.""" + transport = DemoTransport() + writes: list[tuple] = [] + transport.set_units = lambda axis, value: writes.append(("units", axis, value)) + transport.set_speed = lambda axis, value: writes.append(("speed", axis, value)) + transport.set_accel = lambda axis, value: writes.append(("accel", axis, value)) + transport.bas_download = lambda filename, run_mode=1: writes.append(("bas", filename)) + service = MotionService(transport) + + with caplog.at_level(logging.INFO): + result = service.connect(ConnectionSpec(mode="demo")) + + assert result.ok, result.message + assert writes == [] + readback = [record.message for record in caplog.records if record.message.startswith("connect axis 4")] + assert readback, caplog.text + assert "mpos=" in readback[-1] + assert "not written" in caplog.text + + +def test_axis_4_jog_reuses_the_shared_motion_call_and_logs_position(caplog): + service = MotionService(DemoTransport()) + assert service.connect(ConnectionSpec(mode="demo")).ok + service._bus = BusStatus(bus_type="ecat", init_status=INIT_OK, total_axes=4) # noqa: SLF001 + + with caplog.at_level(logging.INFO): + result = service.jog(4, True) + + assert result.ok, result.message + assert result.code == 0 + logged = [record.message for record in caplog.records if record.message.startswith("jog axis 4")] + assert logged, caplog.text + assert "ok=True" in logged[-1] + assert "code=0" in logged[-1] + assert "mpos=" in logged[-1] + + def test_reinit_needs_the_bus_not_to_be_mid_run(): service = MotionService(DemoTransport()) service.connect(ConnectionSpec(mode="demo")) diff --git a/apps/autocal/tests/test_move_buttons.py b/apps/autocal/tests/test_move_buttons.py index aa172ad..94362c9 100644 --- a/apps/autocal/tests/test_move_buttons.py +++ b/apps/autocal/tests/test_move_buttons.py @@ -19,7 +19,7 @@ from __future__ import annotations from orisys_app_shared.i18n import t -from orisys_autocal.motion import BusStatus +from orisys_autocal.motion import AxisState, BusStatus, Snapshot from orisys_autocal.motion.pose_move import plan_pose_move @@ -37,6 +37,44 @@ def _connected(make_window, *, confirm: bool = False): return window +def test_axis_4_controls_stay_available_when_the_bus_count_is_short(make_window): + """A mapped axis 4 uses the same buttons as axes 0-3. + + The bus-count global can still say 4 after the drive is working again. + Type 0 stays disabled, so an axis the controller has not mapped is not + given a button. + """ + window = make_window() + try: + mapped = tuple(AxisState(axis=index, atype=65, read_ok=True) for index in range(5)) + window.axis_panel.update_snapshot( + Snapshot( + connected=True, + bus=BusStatus(bus_type="ecat", init_status=1, total_axes=4), + axes=mapped, + ) + ) + row = window.axis_panel._rows[4] # noqa: SLF001 + assert row["enable_button"].isEnabled() + assert row["fwd_button"].isEnabled() + assert row["rev_button"].isEnabled() + assert row["stop_button"].isEnabled() + + unmapped = mapped[:-1] + (AxisState(axis=4, atype=0, read_ok=True),) + window.axis_panel.update_snapshot( + Snapshot( + connected=True, + bus=BusStatus(bus_type="ecat", init_status=1, total_axes=4), + axes=unmapped, + ) + ) + row = window.axis_panel._rows[4] # noqa: SLF001 + assert not row["fwd_button"].isEnabled() + assert row["stop_button"].isEnabled() is False + finally: + window.close() + + def test_execute_lives_with_the_point_it_executes(make_window): window = make_window() try: diff --git a/apps/autocal/tests/test_pose_move.py b/apps/autocal/tests/test_pose_move.py index ab1ec6a..69e872f 100644 --- a/apps/autocal/tests/test_pose_move.py +++ b/apps/autocal/tests/test_pose_move.py @@ -216,6 +216,63 @@ def test_the_guard_names_an_unknown_direction(): # ------------------------------------------------------------------------ mover +def test_an_execute_plan_moves_the_four_axes_together_and_leaves_z(): + """Calibration execute starts yaw, pitch and X/Y in one command. Z stays put.""" + service = make_service() + called: list[list[tuple[int, float]]] = [] + + def record(moves): + called.append(list(moves)) + return MotionResult.success() + + service.move_abs_many = record # type: ignore[method-assign] + poller = FakePoller(service) + mover = PoseMover(service, poller.submit) + plan = plan_pose_move( + service, + {"x": 2.0, "y": 1.0, "yaw": 10.0, "pitch": -5.0}, + hold=("z",), + together=True, + ) + assert plan.together + assert [step.role for step in plan.steps] == ["yaw", "pitch", "x", "y"] + assert mover.start(plan).ok + assert len(poller.commands) == 1 + poller.commands[0](service) + assert [axis for axis, _value in called[0]] == [3, 2, 0, 1] + + mover.on_snapshot( + snapshot_of(service, {0: 1000.0, 1: 0.0, 2: 0.0, 3: 0.0}, idle=False) + ) + assert mover.running + assert mover.progress_text().startswith("同时") + assert len(poller.commands) == 1, "a later axis was queued instead of moving with the first" + + mover.on_snapshot( + snapshot_of( + service, + { + 0: 1000.0, + 1: 500.0, + 2: -5.0 * 12500.0 / 90.0, + 3: 10.0 * 29500.0 / 90.0, + }, + ) + ) + assert mover.state.finished + + +def test_return_to_zero_stays_one_axis_at_a_time(): + service = make_service() + poller = FakePoller(service) + mover = PoseMover(service, poller.submit) + plan = plan_return_to_initial(service) + assert not plan.together + assert mover.start(plan).ok + assert len(poller.commands) == 1 + assert mover.state.step.role == "z" + + def test_the_mover_commands_one_axis_and_waits_for_it_to_arrive(): service = make_service() poller = FakePoller(service)