vee
119 lines
4.0 KiB
Python
119 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from cognitive_agent.application.calibration import protocol
|
|
from cognitive_agent.application.calibration.service import ApiError, CalibrationService
|
|
|
|
|
|
def test_protocol_matches_android_golden_vectors() -> None:
|
|
assert protocol.crc8(b"123456789") == 0x4B
|
|
vectors = (
|
|
(
|
|
protocol.command(0x00, protocol.OP_HELLO, protocol.AXIS_ALL),
|
|
"A1 00 01 FF 00 00 00 55",
|
|
),
|
|
(
|
|
protocol.command(0x01, protocol.OP_ARM, protocol.AXIS_ALL, 0x4E37),
|
|
"A1 01 02 FF 37 4E 00 0A",
|
|
),
|
|
(
|
|
protocol.command(0x23, protocol.OP_CALIBRATE, protocol.AXIS_ALL),
|
|
"A1 23 10 FF 00 00 00 33",
|
|
),
|
|
(
|
|
protocol.command(0x24, protocol.OP_MOVE_PERMILLE, 0, 1000),
|
|
"A1 24 11 00 E8 03 00 78",
|
|
),
|
|
(
|
|
protocol.command(0x25, protocol.OP_STOP, protocol.AXIS_ALL),
|
|
"A1 25 12 FF 00 00 00 6E",
|
|
),
|
|
)
|
|
for actual, expected in vectors:
|
|
assert actual == bytes.fromhex(expected)
|
|
|
|
|
|
def test_protocol_parses_status_and_rejects_bad_crc() -> None:
|
|
ack = protocol.parse(0x621, bytes.fromhex("B1 22 10 02 00 07 01 A9"))
|
|
assert isinstance(ack, protocol.Ack)
|
|
assert ack.accepted
|
|
|
|
controller = protocol.parse(0x622, bytes.fromhex("C1 01 00 17 07 FF 02 4E"))
|
|
assert isinstance(controller, protocol.ControllerStatus)
|
|
assert controller.can_ready
|
|
assert controller.motor_ready
|
|
assert controller.armed
|
|
|
|
frame = bytearray.fromhex("D1 00 F0 00 E0 01 17 40")
|
|
frame[2] ^= 1
|
|
with pytest.raises(protocol.ProtocolError):
|
|
protocol.parse(0x630, bytes(frame))
|
|
|
|
|
|
def test_simulation_uses_real_wire_boundary_and_generates_pose() -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
service = CalibrationService(Path(directory))
|
|
service.start()
|
|
try:
|
|
initial = service.snapshot()
|
|
assert initial["transport"] == {
|
|
"mode": "hardware",
|
|
"state": "disconnected",
|
|
"detail": "尚未连接",
|
|
}
|
|
|
|
connected = service.connect_can({"mode": "simulation"})
|
|
assert connected["transport"]["state"] == "online"
|
|
assert connected["controller"]["can_ready"]
|
|
assert connected["controller"]["motor_ready"]
|
|
assert [axis["range"] for axis in connected["axes"]] == [480, 647, 917, 373]
|
|
|
|
service.move(0, 800)
|
|
time.sleep(0.2)
|
|
moving = service.snapshot()
|
|
assert moving["controller"]["armed"]
|
|
assert moving["axes"][0]["moving"]
|
|
|
|
time.sleep(0.45)
|
|
pose = service.snapshot()["mqtt"]
|
|
assert pose["state"] == "simulated"
|
|
assert pose["count"] >= 1
|
|
assert "Seat_cushion" in pose["areas"]
|
|
|
|
stopped = service.stop()
|
|
assert not stopped["controller"]["armed"]
|
|
finally:
|
|
service.close()
|
|
|
|
|
|
def test_virtual_session_restart_resets_axes_without_touching_hardware() -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
service = CalibrationService(Path(directory))
|
|
service.start()
|
|
try:
|
|
service.connect_can({"mode": "simulation"})
|
|
service.move(1, 720)
|
|
time.sleep(0.15)
|
|
|
|
reset = service.reset_virtual_session()
|
|
|
|
assert reset["transport"]["mode"] == "simulation"
|
|
assert reset["transport"]["state"] == "disconnected"
|
|
assert reset["controller"]["armed"] is False
|
|
assert [axis["permille"] for axis in reset["axes"]] == [0, 0, 0, 0]
|
|
assert reset["operation"]["phase"] == protocol.PHASE_IDLE
|
|
finally:
|
|
service.close()
|
|
|
|
|
|
def test_virtual_session_restart_rejects_real_bench_state() -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
service = CalibrationService(Path(directory))
|
|
with pytest.raises(ApiError, match="真实台架会话"):
|
|
service.reset_virtual_session()
|