ykcheungab
329 lines
10 KiB
Python
329 lines
10 KiB
Python
"""Guard and state-machine behaviour of the motion service.
|
|
|
|
Driven against the simulated platform, so every assertion here is about AutoCal's
|
|
own logic: what it refuses, what it reads, and when. No test in this file touches
|
|
a controller or a DLL.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from orisys_autocal.motion import (
|
|
INIT_FAILED,
|
|
INIT_OK,
|
|
INIT_RUNNING,
|
|
AxisParams,
|
|
ConnectionSpec,
|
|
DemoTransport,
|
|
MotionService,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def service():
|
|
return MotionService(DemoTransport())
|
|
|
|
|
|
def _initialise(service, timeout: float = 5.0) -> None:
|
|
"""Download a BAS program and wait for the simulated bus to report ready."""
|
|
service.download_bas(__file__, 1)
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline and service.bus.init_status != INIT_OK:
|
|
service.poll([0])
|
|
time.sleep(0.05)
|
|
|
|
|
|
# ----------------------------------------------------------------- connection
|
|
|
|
|
|
def test_a_fresh_service_is_disconnected_and_touches_nothing(service):
|
|
assert not service.connected
|
|
assert not service.bas_loaded
|
|
assert service.poll([0]).connected is False
|
|
|
|
|
|
def test_connect_reports_success_and_an_unknown_bus_without_a_bas(service):
|
|
result = service.connect(ConnectionSpec(mode="demo"))
|
|
assert result.ok
|
|
assert service.connected
|
|
# No BAS program has run, so the bus type must not be guessed at.
|
|
snapshot = service.poll([0])
|
|
assert snapshot.bus.bus_type == "unknown"
|
|
assert not snapshot.bus.initialized
|
|
|
|
|
|
def test_connect_failure_is_reported_without_raising(service):
|
|
result = service.connect(ConnectionSpec(mode="eth", ip="127.0.0.1"))
|
|
assert not result.ok
|
|
assert not service.connected
|
|
|
|
|
|
def test_disconnect_clears_bus_state(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
service.disconnect()
|
|
assert not service.connected
|
|
assert service.bus.bus_type == "unknown"
|
|
|
|
|
|
# ---------------------------------------------------------------- bus init
|
|
|
|
|
|
def test_download_requires_a_real_file(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
assert not service.download_bas("", 1).ok
|
|
assert not service.download_bas(r"D:\nope\missing.bas", 1).ok
|
|
|
|
|
|
def test_download_needs_a_connection(service):
|
|
assert not service.download_bas(__file__, 1).ok
|
|
|
|
|
|
def test_download_then_poll_reaches_initialised(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
assert service.download_bas(__file__, 1).ok
|
|
_initialise(service)
|
|
assert service.bus.init_status == INIT_OK
|
|
assert service.bus.bus_type == "ecat"
|
|
assert service.bus.total_axes == 5
|
|
assert service.bus.node_count == 5
|
|
|
|
|
|
def test_bus_is_running_immediately_after_download(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
service.download_bas(__file__, 1)
|
|
assert service.bus.init_status == INIT_RUNNING
|
|
|
|
|
|
def test_reinitialise_is_refused_while_already_running(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
service.download_bas(__file__, 1)
|
|
assert not service.reinit_bus().ok
|
|
|
|
|
|
def test_reinitialise_is_refused_without_a_bas_program(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
result = service.reinit_bus()
|
|
assert not result.ok
|
|
assert "BAS" in result.message or "Bas" in result.message
|
|
|
|
|
|
def test_init_timeout_is_reported_rather_than_raising():
|
|
"""A controller that never publishes its globals must time out, not hang."""
|
|
|
|
class SilentTransport(DemoTransport):
|
|
def user_var(self, name):
|
|
from orisys_app_shared.zmotion import ZAuxError
|
|
|
|
raise ZAuxError(30002, f"{name} is not defined")
|
|
|
|
service = MotionService(SilentTransport())
|
|
service._init_timeout = 0.05 # noqa: SLF001 - shortening the vendor window
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
service.download_bas(__file__, 1)
|
|
time.sleep(0.1)
|
|
snapshot = service.poll([0])
|
|
assert snapshot.bus.init_status == INIT_FAILED
|
|
assert snapshot.bus.message # carries the ZDevelop hint, not an empty string
|
|
|
|
|
|
# ------------------------------------------------------------------- guards
|
|
|
|
|
|
def test_motion_is_refused_before_the_bus_is_initialised(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
for attempt in (
|
|
service.jog(0, True),
|
|
service.move_abs(0, 100.0),
|
|
service.move_rel(0, 10.0),
|
|
service.home(0),
|
|
):
|
|
assert not attempt.ok
|
|
assert attempt.message
|
|
|
|
|
|
def test_motion_is_refused_when_not_connected(service):
|
|
assert not service.jog(0, True).ok
|
|
assert not service.move_abs(0, 1.0).ok
|
|
assert not service.stop(0).ok
|
|
|
|
|
|
def test_motion_on_an_unknown_axis_is_refused(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
_initialise(service)
|
|
result = service.jog(99, True)
|
|
assert not result.ok
|
|
assert "99" in result.message or "0-4" in result.message
|
|
|
|
|
|
@pytest.mark.parametrize("bad", [float("nan"), float("inf"), float("-inf")])
|
|
def test_a_non_finite_target_is_refused(service, bad):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
_initialise(service)
|
|
result = service.move_abs(0, bad)
|
|
assert not result.ok
|
|
assert result.message
|
|
|
|
|
|
def test_home_needs_a_configured_mode(service):
|
|
"""0 is not a valid homing mode; inheriting one would be worse than saying so."""
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
_initialise(service)
|
|
result = service.home(0)
|
|
assert not result.ok
|
|
assert "0" in result.message
|
|
|
|
|
|
def test_home_runs_once_a_mode_is_set(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
_initialise(service)
|
|
service.set_axis_params(0, AxisParams(home_mode=17))
|
|
assert service.home(0).ok
|
|
|
|
|
|
def test_home_is_refused_on_rtex(service):
|
|
"""The vendor example refuses RTEX homing; so do we."""
|
|
|
|
class RtexTransport(DemoTransport):
|
|
def user_var(self, name):
|
|
return 1.0 if name == "BUS_TYPE" else super().user_var(name)
|
|
|
|
rtex = MotionService(RtexTransport())
|
|
rtex.connect(ConnectionSpec(mode="demo"))
|
|
_initialise(rtex)
|
|
assert rtex.bus.bus_type == "rtex"
|
|
result = rtex.home(0)
|
|
assert not result.ok
|
|
assert result.message
|
|
|
|
|
|
# ------------------------------------------------------------------- motion
|
|
|
|
|
|
def test_jog_and_stop_move_the_state_machine(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
_initialise(service)
|
|
assert service.jog(0, True).ok
|
|
state = service.poll([0]).axis(0)
|
|
assert state is not None and not state.idle
|
|
|
|
assert service.stop(0).ok
|
|
state = service.poll([0]).axis(0)
|
|
assert state is not None and state.idle
|
|
|
|
|
|
def test_absolute_and_relative_moves_change_the_command_position(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
_initialise(service)
|
|
assert service.move_abs(1, 250.0).ok
|
|
assert service.poll([1]).axis(1).dpos == pytest.approx(250.0)
|
|
assert service.move_rel(1, -50.0).ok
|
|
assert service.poll([1]).axis(1).dpos == pytest.approx(200.0)
|
|
|
|
|
|
def test_set_axis_params_writes_and_remembers(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
_initialise(service)
|
|
params = AxisParams(units=5.0, speed=200.0, accel=2000.0)
|
|
assert service.set_axis_params(2, params).ok
|
|
assert service.params(2) == params
|
|
assert service.to_mm(2, 1000.0) == pytest.approx(1.0)
|
|
|
|
|
|
def test_enable_toggles_the_axis(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
_initialise(service)
|
|
assert service.set_axis_enable(3, False).ok
|
|
assert service.poll([3]).axis(3).enabled is False
|
|
assert service.set_axis_enable(3, True).ok
|
|
assert service.poll([3]).axis(3).enabled is True
|
|
|
|
|
|
def test_stop_all_covers_every_axis(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
_initialise(service)
|
|
results = service.stop_all()
|
|
assert len(results) == service.axis_count
|
|
assert all(result.ok for result in results)
|
|
|
|
|
|
def test_clear_alarm_reaches_the_drive(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
_initialise(service)
|
|
assert service.clear_alarm(0).ok
|
|
|
|
|
|
# ----------------------------------------------------------------- bus data
|
|
|
|
|
|
def test_sdo_round_trips_on_ethercat(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
_initialise(service)
|
|
assert service.sdo_write(0, 0x6040, 0, 3, 6).ok
|
|
result, value = service.sdo_read(0, 0x6040, 0, 3)
|
|
assert result.ok
|
|
assert value == 6
|
|
|
|
|
|
def test_sdo_is_refused_on_rtex(service):
|
|
class RtexTransport(DemoTransport):
|
|
def user_var(self, name):
|
|
return 1.0 if name == "BUS_TYPE" else super().user_var(name)
|
|
|
|
rtex = MotionService(RtexTransport())
|
|
rtex.connect(ConnectionSpec(mode="demo"))
|
|
_initialise(rtex)
|
|
result, _value = rtex.sdo_read(0, 0x6040, 0, 3)
|
|
assert not result.ok
|
|
|
|
|
|
def test_rtex_is_refused_on_ethercat(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
_initialise(service)
|
|
result, _value = service.rtex_read(0, 1)
|
|
assert not result.ok
|
|
|
|
|
|
def test_bus_data_is_refused_before_the_bus_type_is_known(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
result, _value = service.sdo_read(0, 0x6040, 0, 3)
|
|
assert not result.ok
|
|
|
|
|
|
# ------------------------------------------------------------------- polling
|
|
|
|
|
|
def test_polling_reports_every_requested_axis(service):
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
snapshot = service.poll([0, 1, 2, 3, 4])
|
|
assert [state.axis for state in snapshot.axes] == [0, 1, 2, 3, 4]
|
|
assert all(state.read_ok for state in snapshot.axes)
|
|
|
|
|
|
def test_a_read_failure_is_marked_not_silently_zeroed():
|
|
"""A failed read must not look like a position of zero."""
|
|
|
|
class FailingReads(DemoTransport):
|
|
def get_dpos(self, axis):
|
|
from orisys_app_shared.zmotion import ZAuxError
|
|
|
|
raise ZAuxError(30000, "no response")
|
|
|
|
service = MotionService(FailingReads())
|
|
service.connect(ConnectionSpec(mode="demo"))
|
|
snapshot = service.poll([0])
|
|
state = snapshot.axis(0)
|
|
assert state is not None
|
|
assert state.read_ok is False
|
|
assert state.error
|
|
|
|
|
|
def test_language_switches_the_refusal_text(service):
|
|
assert not service.jog(0, True).ok
|
|
zh = service.jog(0, True).message
|
|
service.set_language("en_US")
|
|
en = service.jog(0, True).message
|
|
assert zh != en
|