"""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 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()