296 lines
11 KiB
Python
296 lines
11 KiB
Python
"""Laser-control protocol compatibility tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from python_app.hardware_full.laser_control.controller import DEVICE_MAIN_MESSAGE_ID, LaserController
|
|
from python_app.hardware_full.laser_control.models import VariationType
|
|
from python_app.hardware_full.laser_control.protocol import Protocol, TaskType
|
|
from python_app.hardware_full.kamil_adc_service import apply_kamil_adc_laser_control
|
|
from python_app.models.run_config_model import RunConfigModel
|
|
|
|
|
|
DEVICE_MAIN_MANUAL_HEX = "1111ff37ffa518ab000000000000000a8000000a8000ff003d2acc2c163f"
|
|
DEVICE_MAIN_CHANGE_CURRENT_LD1_HEX = (
|
|
"7777ff3701003d2acc2c10008813ffa5cc2c18ab0a00000a8000000a8000b600"
|
|
)
|
|
DEVICE_MAIN_CHANGE_CURRENT_LD2_HEX = (
|
|
# Word 5 (step) = current_ma_to_n(0.05)=0x0010, matching LD1 and min/max — see the
|
|
# step-encoding fix in protocol.py (was the inconsistent int(step*100)=0x0005).
|
|
"7777ff3702003d2acc2c1000881318ab3d2affa50a00000a8000000a80004406"
|
|
)
|
|
|
|
|
|
class _FakeProtocol:
|
|
def __init__(self) -> None:
|
|
self.is_connected = True
|
|
self.sent: list[bytes] = []
|
|
|
|
def send_raw(self, data: bytes) -> None:
|
|
self.sent.append(bytes(data))
|
|
|
|
def receive_raw(self, length: int) -> bytes:
|
|
return b"\x00\x00" if length == 2 else b""
|
|
|
|
|
|
class _FakeLaserController:
|
|
instances: list["_FakeLaserController"] = []
|
|
|
|
def __init__(self, **kwargs: object) -> None:
|
|
self.kwargs = kwargs
|
|
self.calls: list[tuple[str, object]] = []
|
|
_FakeLaserController.instances.append(self)
|
|
|
|
def connect(self) -> bool:
|
|
self.calls.append(("connect", None))
|
|
return True
|
|
|
|
def reset(self) -> None:
|
|
self.calls.append(("reset", None))
|
|
|
|
def set_manual_mode(self, **kwargs: object) -> None:
|
|
self.calls.append(("set_manual_mode", kwargs))
|
|
|
|
def start_variation(self, **kwargs: object) -> None:
|
|
self.calls.append(("start_variation", kwargs))
|
|
|
|
def disconnect(self) -> None:
|
|
self.calls.append(("disconnect", None))
|
|
|
|
|
|
class LaserControlProtocolCompatibilityTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
_FakeLaserController.instances.clear()
|
|
|
|
@staticmethod
|
|
def _kamil_config(laser_payload: dict[str, object]) -> RunConfigModel:
|
|
return RunConfigModel.from_dict(
|
|
{
|
|
"radar": {
|
|
"model": "kamil_adc",
|
|
"driver_mode": "native",
|
|
"laser_control": laser_payload,
|
|
}
|
|
}
|
|
)
|
|
|
|
def test_manual_command_matches_device_main_bytes(self) -> None:
|
|
command = Protocol.encode_decode_enable(
|
|
temp1=28.0,
|
|
temp2=28.9,
|
|
current1=33.0,
|
|
current2=35.0,
|
|
pi_coeff1_p=2560,
|
|
pi_coeff1_i=128,
|
|
pi_coeff2_p=2560,
|
|
pi_coeff2_i=128,
|
|
message_id=DEVICE_MAIN_MESSAGE_ID,
|
|
)
|
|
|
|
self.assertEqual(command.hex(), DEVICE_MAIN_MANUAL_HEX)
|
|
|
|
def test_variation_commands_match_device_main_bytes(self) -> None:
|
|
ld1_command = Protocol.encode_task_enable(
|
|
task_type=TaskType.CHANGE_CURRENT_LD1,
|
|
static_temp1=28.0,
|
|
static_temp2=28.9,
|
|
static_current1=33.0,
|
|
static_current2=35.0,
|
|
min_value=33.0,
|
|
max_value=35.0,
|
|
step=0.05,
|
|
time_step=50,
|
|
delay_time=10,
|
|
message_id=DEVICE_MAIN_MESSAGE_ID,
|
|
pi_coeff1_p=2560,
|
|
pi_coeff1_i=128,
|
|
pi_coeff2_p=2560,
|
|
pi_coeff2_i=128,
|
|
)
|
|
ld2_command = Protocol.encode_task_enable(
|
|
task_type=TaskType.CHANGE_CURRENT_LD2,
|
|
static_temp1=28.0,
|
|
static_temp2=28.9,
|
|
static_current1=33.0,
|
|
static_current2=35.0,
|
|
min_value=33.0,
|
|
max_value=35.0,
|
|
step=0.05,
|
|
time_step=50,
|
|
delay_time=10,
|
|
message_id=DEVICE_MAIN_MESSAGE_ID,
|
|
pi_coeff1_p=2560,
|
|
pi_coeff1_i=128,
|
|
pi_coeff2_p=2560,
|
|
pi_coeff2_i=128,
|
|
)
|
|
|
|
self.assertEqual(ld1_command.hex(), DEVICE_MAIN_CHANGE_CURRENT_LD1_HEX)
|
|
self.assertEqual(ld2_command.hex(), DEVICE_MAIN_CHANGE_CURRENT_LD2_HEX)
|
|
|
|
def test_ld1_and_ld2_encode_current_step_identically(self) -> None:
|
|
# Regression guard for the LD2 step-scale bug: both current-variation channels
|
|
# must encode the step with the same scale (current_ma_to_n), like min/max.
|
|
params = dict(static_temp1=28.0, static_temp2=28.9, static_current1=33.0, static_current2=35.0,
|
|
min_value=33.0, max_value=35.0, step=0.05, time_step=50, delay_time=10,
|
|
message_id=DEVICE_MAIN_MESSAGE_ID, pi_coeff1_p=2560, pi_coeff1_i=128,
|
|
pi_coeff2_p=2560, pi_coeff2_i=128)
|
|
ld1 = Protocol.encode_task_enable(task_type=TaskType.CHANGE_CURRENT_LD1, **params)
|
|
ld2 = Protocol.encode_task_enable(task_type=TaskType.CHANGE_CURRENT_LD2, **params)
|
|
# Word 5 (step) is at bytes 10:12 in both frames (sync, header, task, min, max, step).
|
|
self.assertEqual(ld1[10:12], ld2[10:12])
|
|
|
|
def test_start_sequence_matches_device_main_order(self) -> None:
|
|
fake_protocol = _FakeProtocol()
|
|
controller = LaserController(pi_coeff1_p=2560, pi_coeff1_i=128, pi_coeff2_p=2560, pi_coeff2_i=128)
|
|
controller._protocol = fake_protocol
|
|
|
|
with patch("python_app.hardware_full.laser_control.controller.time.sleep", return_value=None):
|
|
controller.reset()
|
|
controller.set_manual_mode(
|
|
temp1=28.0,
|
|
temp2=28.9,
|
|
current1=33.0,
|
|
current2=35.0,
|
|
message_id=DEVICE_MAIN_MESSAGE_ID,
|
|
)
|
|
controller.start_variation(
|
|
variation_type=VariationType.CHANGE_CURRENT_LD1,
|
|
params={
|
|
"static_temp1": 28.0,
|
|
"static_temp2": 28.9,
|
|
"static_current1": 33.0,
|
|
"static_current2": 35.0,
|
|
"min_value": 33.0,
|
|
"max_value": 35.0,
|
|
"step": 0.05,
|
|
"time_step": 50,
|
|
"delay_time": 10,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(
|
|
[command.hex() for command in fake_protocol.sent],
|
|
[
|
|
"2222",
|
|
DEVICE_MAIN_MANUAL_HEX,
|
|
DEVICE_MAIN_CHANGE_CURRENT_LD1_HEX,
|
|
],
|
|
)
|
|
|
|
def test_stop_sequence_restores_device_main_manual_bytes(self) -> None:
|
|
fake_protocol = _FakeProtocol()
|
|
controller = LaserController(pi_coeff1_p=2560, pi_coeff1_i=128, pi_coeff2_p=2560, pi_coeff2_i=128)
|
|
controller._protocol = fake_protocol
|
|
|
|
with patch("python_app.hardware_full.laser_control.controller.time.sleep", return_value=None):
|
|
controller.set_manual_mode(
|
|
temp1=28.0,
|
|
temp2=28.9,
|
|
current1=33.0,
|
|
current2=35.0,
|
|
message_id=DEVICE_MAIN_MESSAGE_ID,
|
|
)
|
|
fake_protocol.sent.clear()
|
|
controller.stop_task(restore_message_id=DEVICE_MAIN_MESSAGE_ID)
|
|
|
|
self.assertEqual(
|
|
[command.hex() for command in fake_protocol.sent],
|
|
[
|
|
"2222",
|
|
DEVICE_MAIN_MANUAL_HEX,
|
|
],
|
|
)
|
|
|
|
def test_apply_radar_variation_sequence_matches_device_main_order(self) -> None:
|
|
config = self._kamil_config(
|
|
{
|
|
"enabled": True,
|
|
"port": "/dev/ttyUSB0",
|
|
"mode": "variation",
|
|
"pi_coeff1_p": 2560,
|
|
"pi_coeff1_i": 128,
|
|
"pi_coeff2_p": 2560,
|
|
"pi_coeff2_i": 128,
|
|
"variation": {
|
|
"variation_type": "CHANGE_CURRENT_LD1",
|
|
"static_temp1": 28.0,
|
|
"static_temp2": 28.9,
|
|
"static_current1": 33.0,
|
|
"static_current2": 35.0,
|
|
"min_value": 33.0,
|
|
"max_value": 35.0,
|
|
"step": 0.05,
|
|
"time_step": 50,
|
|
"delay_time": 10,
|
|
},
|
|
}
|
|
)
|
|
|
|
with patch("python_app.hardware_full.laser_control.controller.LaserController", _FakeLaserController):
|
|
applied = apply_kamil_adc_laser_control(config)
|
|
|
|
self.assertTrue(applied)
|
|
controller = _FakeLaserController.instances[0]
|
|
self.assertEqual(
|
|
[name for name, _payload in controller.calls],
|
|
["connect", "reset", "set_manual_mode", "start_variation", "disconnect"],
|
|
)
|
|
manual_payload = controller.calls[2][1]
|
|
self.assertEqual(
|
|
manual_payload,
|
|
{
|
|
"temp1": 28.0,
|
|
"temp2": 28.9,
|
|
"current1": 33.0,
|
|
"current2": 35.0,
|
|
"message_id": DEVICE_MAIN_MESSAGE_ID,
|
|
},
|
|
)
|
|
# The variation handoff itself (type + params), not just call order, must be right.
|
|
variation_payload = controller.calls[3][1]
|
|
self.assertEqual(variation_payload["variation_type"], VariationType.CHANGE_CURRENT_LD1)
|
|
self.assertEqual(variation_payload["params"]["step"], 0.05)
|
|
self.assertEqual(variation_payload["params"]["min_value"], 33.0)
|
|
self.assertEqual(variation_payload["params"]["max_value"], 35.0)
|
|
|
|
def test_apply_radar_manual_mode_sets_manual_without_variation(self) -> None:
|
|
config = self._kamil_config(
|
|
{
|
|
"enabled": True,
|
|
"port": "/dev/ttyUSB0",
|
|
"mode": "manual",
|
|
"manual": {"temp1": 26.0, "temp2": 27.0, "current1": 30.0, "current2": 31.0},
|
|
}
|
|
)
|
|
with patch("python_app.hardware_full.laser_control.controller.LaserController", _FakeLaserController):
|
|
applied = apply_kamil_adc_laser_control(config)
|
|
self.assertTrue(applied)
|
|
controller = _FakeLaserController.instances[0]
|
|
self.assertEqual([name for name, _ in controller.calls], ["connect", "reset", "set_manual_mode", "disconnect"])
|
|
self.assertEqual(controller.calls[2][1]["current1"], 30.0)
|
|
|
|
def test_apply_radar_unknown_variation_type_raises(self) -> None:
|
|
config = self._kamil_config(
|
|
{"enabled": True, "port": "/dev/ttyUSB0", "mode": "variation",
|
|
"variation": {"variation_type": "NOT_A_REAL_TYPE"}}
|
|
)
|
|
with patch("python_app.hardware_full.laser_control.controller.LaserController", _FakeLaserController):
|
|
with self.assertRaisesRegex(ValueError, "variation_type"):
|
|
apply_kamil_adc_laser_control(config)
|
|
|
|
def test_apply_radar_skips_disabled_laser_control(self) -> None:
|
|
config = self._kamil_config({"enabled": False})
|
|
|
|
with patch("python_app.hardware_full.laser_control.controller.LaserController", _FakeLaserController):
|
|
applied = apply_kamil_adc_laser_control(config)
|
|
|
|
self.assertFalse(applied)
|
|
self.assertEqual(_FakeLaserController.instances, [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|