ykcheungab
b4c5d65730
保留已确认配置下的轴号、目标数值和发送顺序，拒绝不匹配的单位或过期计划。136 项相关检查通过，8256 条运动调用与修改前一致；全套既有失败见运动单位契约。
188 lines
6.6 KiB
Python
188 lines
6.6 KiB
Python
"""Keep commissioned command numbers; reject unverified physical-unit moves."""
|
|
|
|
from dataclasses import replace
|
|
import math
|
|
|
|
import pytest
|
|
|
|
from orisys_autocal.motion.models import AxisParams, BusStatus, ConnectionSpec
|
|
from orisys_autocal.motion.pose_move import PoseMover, plan_pose_move
|
|
from orisys_autocal.motion.service import MotionService
|
|
from orisys_autocal.motion.transport import DemoTransport
|
|
from orisys_autocal.motion.units import deg_to_user_units, user_units_to_deg
|
|
from orisys_app_shared.zmotion import ZAuxError
|
|
|
|
|
|
class RecordingTransport(DemoTransport):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.moves = []
|
|
|
|
def move_abs(self, axis, value):
|
|
self.moves.append((axis, value))
|
|
return super().move_abs(axis, value)
|
|
|
|
|
|
def service():
|
|
transport = RecordingTransport()
|
|
instance = MotionService(
|
|
transport,
|
|
language="en_US",
|
|
limits={"max_travel_mm": 100.0, "max_yaw_deg": 180.0, "max_pitch_deg": 90.0},
|
|
)
|
|
instance.connect(ConnectionSpec(mode="demo"))
|
|
instance.poll()
|
|
instance._bus = BusStatus(init_status=1, total_axes=5)
|
|
return instance, transport
|
|
|
|
|
|
def mover_and_actions(instance, targets):
|
|
actions = []
|
|
|
|
def submit(action):
|
|
actions.append(action)
|
|
return len(actions)
|
|
|
|
mover = PoseMover(instance, submit)
|
|
plan = plan_pose_move(instance, targets, together=True)
|
|
assert mover.start(plan).ok
|
|
return mover, actions
|
|
|
|
|
|
def test_commissioned_command_numbers_and_order_remain_exact(qt_app):
|
|
instance, transport = service()
|
|
mover, actions = mover_and_actions(
|
|
instance, {"x": -10.0, "y": 2.0, "yaw": 90.0, "pitch": -45.0, "z": 3.0}
|
|
)
|
|
assert actions[0](instance).ok
|
|
assert transport.moves == [
|
|
(2, 12500.0),
|
|
(3, -14750.0),
|
|
(0, -5000.0),
|
|
(1, 1000.0),
|
|
(4, 1500.0),
|
|
]
|
|
|
|
|
|
def test_readback_mismatch_refuses_instead_of_changing_the_command():
|
|
instance, transport = service()
|
|
transport.set_units(0, 5.0)
|
|
instance.poll([0])
|
|
instance._bus = BusStatus(init_status=1, total_axes=5)
|
|
refusal = instance.check_axes_move({"x": 2.0})
|
|
assert refusal is not None and "UNITS" in refusal.message
|
|
assert math.isnan(instance.to_mm(0, 1000.0))
|
|
assert transport.moves == []
|
|
|
|
|
|
def test_explicit_matching_linear_units_are_supported_without_rescaling_other_axes():
|
|
instance, transport = service()
|
|
assert instance.set_axis_params(0, AxisParams(units=5.0)).ok
|
|
instance.poll([0])
|
|
instance._bus = BusStatus(init_status=1, total_axes=5)
|
|
assert instance.check_axes_move({"x": 2.0}) is None
|
|
assert instance.to_units("x", 2.0) == 2000.0
|
|
assert instance.to_mm(0, 1000.0) == 1.0
|
|
assert instance.to_units("yaw", 90.0) == 12500.0
|
|
|
|
|
|
@pytest.mark.parametrize("units", [0.0, -1.0, math.nan, math.inf])
|
|
def test_invalid_unit_readback_is_not_a_physical_measurement(units):
|
|
instance, transport = service()
|
|
transport.set_units(0, units)
|
|
instance.poll([0])
|
|
instance._bus = BusStatus(init_status=1, total_axes=5)
|
|
assert instance.check_axes_move({"x": 2.0}) is not None
|
|
assert math.isnan(instance.to_mm(0, 1000.0))
|
|
|
|
|
|
def test_missing_or_failed_readback_cannot_reuse_the_last_good_scale():
|
|
instance, transport = service()
|
|
instance._last_axes.clear()
|
|
assert instance.check_axes_move({"x": 2.0}) is not None
|
|
instance.poll([0])
|
|
instance._last_axes[0] = replace(instance._last_axes[0], read_ok=False)
|
|
assert instance.check_axes_move({"x": 2.0}) is not None
|
|
assert math.isnan(instance.to_mm(0, 1000.0))
|
|
|
|
|
|
@pytest.mark.parametrize("axis,role", [(2, "yaw"), (3, "pitch")])
|
|
def test_changed_rotary_pulse_equivalent_requires_new_calibration(axis, role):
|
|
instance, transport = service()
|
|
assert instance.set_axis_params(axis, AxisParams(units=5.0)).ok
|
|
instance.poll([axis])
|
|
instance._bus = BusStatus(init_status=1, total_axes=5)
|
|
refusal = instance.check_axes_move({role: 45.0})
|
|
assert refusal is not None and "calibrat" in refusal.message
|
|
assert math.isnan(instance.to_units(role, 45.0))
|
|
assert math.isnan(instance.to_deg(axis, 1000.0))
|
|
assert math.isnan(user_units_to_deg(axis, 1000.0, units=5.0))
|
|
assert math.isnan(deg_to_user_units(axis, 45.0, units=5.0))
|
|
|
|
|
|
def test_change_after_queueing_is_refused_before_any_axis_moves(qt_app):
|
|
instance, transport = service()
|
|
mover, actions = mover_and_actions(instance, {"x": 2.0, "yaw": 10.0})
|
|
# No poll between the external parameter change and worker execution.
|
|
transport.set_units(0, 5.0)
|
|
result = actions[0](instance)
|
|
assert not result.ok
|
|
mover.on_command_finished(1, result)
|
|
assert not mover.running and "UNITS" in mover.state.failure
|
|
assert transport.moves == []
|
|
|
|
|
|
def test_unrelated_command_result_does_not_cancel_a_pose_move(qt_app):
|
|
from orisys_autocal.motion.models import MotionResult
|
|
|
|
instance, transport = service()
|
|
mover, actions = mover_and_actions(instance, {"x": 2.0})
|
|
mover.on_command_finished(99, MotionResult.refused("another command"))
|
|
assert mover.running
|
|
assert actions[0](instance).ok
|
|
mover.on_command_failed(1, "read failed")
|
|
assert not mover.running and mover.state.failure == "read failed"
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["parameters", "mapping"])
|
|
def test_a_queued_plan_cannot_be_reinterpreted_under_new_settings(qt_app, change):
|
|
instance, transport = service()
|
|
mover, actions = mover_and_actions(instance, {"x": 2.0, "y": 1.0})
|
|
if change == "parameters":
|
|
instance.set_axis_params(0, AxisParams(units=5.0))
|
|
else:
|
|
instance.set_axis_roles({"x": 1, "y": 0, "yaw": 2, "pitch": 3, "z": 4})
|
|
assert not actions[0](instance).ok
|
|
assert transport.moves == []
|
|
|
|
|
|
def test_a_stale_plan_is_refused_before_it_is_queued(qt_app):
|
|
instance, transport = service()
|
|
plan = plan_pose_move(instance, {"x": 2.0}, together=True)
|
|
instance.set_axis_params(0, AxisParams(units=5.0))
|
|
instance.poll([0])
|
|
instance._bus = BusStatus(init_status=1, total_axes=5)
|
|
actions = []
|
|
mover = PoseMover(instance, lambda action: actions.append(action))
|
|
assert not mover.start(plan).ok
|
|
assert actions == [] and transport.moves == []
|
|
|
|
|
|
def test_unit_read_error_before_execution_refuses_the_entire_group(qt_app):
|
|
instance, transport = service()
|
|
mover, actions = mover_and_actions(instance, {"x": 2.0, "yaw": 10.0})
|
|
|
|
def unreadable(axis):
|
|
raise ZAuxError(30500, "unit read failed")
|
|
|
|
transport.get_units = unreadable
|
|
assert not actions[0](instance).ok
|
|
assert transport.moves == []
|
|
|
|
|
|
def test_raw_debug_moves_keep_the_existing_user_unit_contract():
|
|
instance, transport = service()
|
|
transport.set_units(2, 5.0)
|
|
assert instance.move_abs(2, 90.0).ok
|
|
assert transport.moves == [(2, 90.0)]
|