orisys-tech/orisys-app
3
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

Z轴可利用代码控制移动

Browse Source
This commit is contained in:
yijian
2026-10-10 14:10:59 +08:00
parent b827c5ffee
commit 264275b1e6
9 changed files with 406 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/autocal/config/locales/en_US.json
+2 -2
View File
@@ -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",
apps/autocal/config/locales/zh_CN.json
+2 -2
View File
@@ -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": "已取消，未发出任何运动",
apps/autocal/src/orisys_autocal/motion/pose_move.py
+79 -6
View File
@@ -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 ""
apps/autocal/src/orisys_autocal/motion/service.py
+152 -17
View File
@@ -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(
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,15 +831,59 @@ 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(
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
apps/autocal/src/orisys_autocal/ui/axis_panel.py
+14 -1
View File
@@ -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."""
apps/autocal/src/orisys_autocal/ui/window.py
+16 -6
View File
@@ -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" {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:
apps/autocal/tests/test_controller_sequence.py
+40
View File
@@ -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"))
apps/autocal/tests/test_move_buttons.py
+39 -1
View File
@@ -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:
apps/autocal/tests/test_pose_move.py
+57
View File
@@ -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)