"""Laser-control protocol compatibility tests.""" from __future__ import annotations from contextlib import ExitStack import unittest from unittest.mock import Mock, patch from python_app.hardware_full.laser_control.controller import DEVICE_MAIN_MESSAGE_ID, LaserController from python_app.hardware_full.laser_control.exceptions import CommunicationError, ValidationError from python_app.hardware_full.laser_control.protocol import Protocol from python_app.hardware_full.laser_control.conversions import variation_step_count from python_app.hardware_full.rph_radar.laser import apply_rph_radar_laser_control from python_app.models.run_config_schema import RunConfigModel # Generated by the lab GUI encoder (RadioPhotonic_PCB_PC_software device_commands) for these inputs. DEVICE_MAIN_MANUAL_HEX = "1111ff37ffa518ab000000000000000a8000000a8000ff003d2acd2c173f" DEVICE_MAIN_CHANGE_CURRENT_LD1_HEX = ( "7777ff3701003d2acd2c10008813ffa5cd2c18ab0a00000a8000000a8000b600" ) 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 LaserControlProtocolCompatibilityTest(unittest.TestCase): def setUp(self) -> None: self.controller = Mock(spec=LaserController) self.controller.connect.return_value = True @staticmethod def _rph_config(laser_payload: dict[str, object]) -> RunConfigModel: return RunConfigModel.from_dict( { "radar": { "model": "rph_radar", "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( 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, 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) def test_sweep_step_count_follows_the_firmware_loop(self) -> None: self.assertEqual(variation_step_count(97.3, 107, 0.05), 199) # 199 steps seen on the RFG stand def test_start_sequence_matches_device_main_order(self) -> None: cleanup = ExitStack() self.addCleanup(cleanup.close) fake_protocol = _FakeProtocol() controller = LaserController("/dev/ttyLASER", pi_coeff1_p=2560, pi_coeff1_i=128, pi_coeff2_p=2560, pi_coeff2_i=128) cleanup.enter_context(patch.object(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( 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_apply_radar_variation_sequence_matches_device_main_order(self) -> None: config = self._rph_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.rph_radar.laser.LaserController", return_value=self.controller): applied = apply_rph_radar_laser_control(config) self.assertTrue(applied) controller = self.controller self.assertEqual( [call[0] for call in controller.method_calls], ["connect", "reset", "set_manual_mode", "start_variation", "disconnect"], ) manual_payload = controller.set_manual_mode.call_args.kwargs 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.start_variation.call_args.kwargs 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._rph_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.rph_radar.laser.LaserController", return_value=self.controller): applied = apply_rph_radar_laser_control(config) self.assertTrue(applied) controller = self.controller self.assertEqual([call[0] for call in controller.method_calls], ["connect", "reset", "set_manual_mode", "disconnect"]) self.assertEqual(controller.set_manual_mode.call_args.kwargs["current1"], 30.0) def test_apply_radar_unknown_variation_type_raises(self) -> None: config = self._rph_config( {"enabled": True, "port": "/dev/ttyUSB0", "mode": "variation", "variation": {"variation_type": "NOT_A_REAL_TYPE"}} ) with patch("python_app.hardware_full.rph_radar.laser.LaserController", return_value=self.controller): with self.assertRaisesRegex(ValueError, "variation_type"): apply_rph_radar_laser_control(config) def test_state_failure_cannot_report_command_success(self) -> None: controller = LaserController("/dev/ttyLASER", pi_coeff1_p=1280, pi_coeff1_i=25, pi_coeff2_p=1280, pi_coeff2_i=179) for reply in (b"", b"\x00", b"\x04\x00", b"\x03\x00"): with self.subTest(reply=reply), patch.object(controller, "_protocol") as protocol: protocol.receive_raw.return_value = reply with patch("python_app.hardware_full.laser_control.controller.time.sleep"): with self.assertRaises(CommunicationError): controller.reset() def test_invalid_settings_are_rejected_before_connect_or_reset(self) -> None: for mode, field, value in (("manual", "temp1", 100.0), ("variation", "static_current2", 150.0), ("variation", "step", 0.0), ("variation", "max_value", 97.3001), # same DAC code as min ("variation", "variation_type", "CHANGE_TEMPERATURE_LD1"), ("pi", "pi_coeff1_p", 65536)): config = self._rph_config({"enabled": True, "port": "/dev/ttyUSB0", "mode": "manual" if mode == "pi" else mode}) target = config.radar.laser_control setattr(target if mode == "pi" else getattr(target, mode), field, value) with self.subTest(mode=mode, field=field), patch( "python_app.hardware_full.rph_radar.laser.LaserController", return_value=self.controller ) as factory: with self.assertRaises((ValueError, ValidationError)): apply_rph_radar_laser_control(config) factory.assert_not_called() def test_apply_radar_skips_disabled_laser_control(self) -> None: config = self._rph_config({"enabled": False}) with patch("python_app.hardware_full.rph_radar.laser.LaserController", return_value=self.controller) as factory: applied = apply_rph_radar_laser_control(config) self.assertFalse(applied) factory.assert_not_called() if __name__ == "__main__": unittest.main()