"""Tests for Kamil ADC config, parser, and producer wiring.""" from __future__ import annotations import json import os from pathlib import Path import pty import struct import sys import tempfile import tty import unittest from python_app.hardware_full.kamil_adc_service import KamilAdcFrameParser, KamilAdcTtyReader from python_app.models.run_config_model import RunConfigModel from python_app.orchestration.process_supervisor import ProcessSupervisor def _start_frame() -> bytes: return struct.pack(" bytes: return struct.pack(" None: value = KamilAdcFrameParser.parse_point(_point_frame(1, 123, -45), expected_step=1) self.assertEqual(value, complex(123, -45)) def test_bad_marker_is_rejected(self) -> None: with self.assertRaisesRegex(ValueError, "marker mismatch"): KamilAdcFrameParser.parse_point(_point_frame(1, 10, 20, marker=0x001A), expected_step=1) def test_wrong_step_is_rejected(self) -> None: with self.assertRaisesRegex(ValueError, "step mismatch"): KamilAdcFrameParser.parse_point(_point_frame(2, 10, 20), expected_step=1) class KamilAdcTtyReaderTest(unittest.TestCase): def test_valid_stream_reads_complex_sweep(self) -> None: master_fd, slave_fd = pty.openpty() reader: KamilAdcTtyReader | None = None try: tty.setraw(slave_fd) reader = KamilAdcTtyReader(os.ttyname(slave_fd)) reader.open() os.write( master_fd, _start_frame() + _point_frame(1, 10, -1) + _point_frame(2, -20, 2) + _start_frame(), ) values = reader.read_sweep(timeout_s=1.0) self.assertEqual(values.tolist(), [complex(10, -1), complex(-20, 2)]) finally: if reader is not None: reader.close() os.close(master_fd) os.close(slave_fd) def test_stream_reads_consecutive_variable_length_sweeps(self) -> None: master_fd, slave_fd = pty.openpty() reader: KamilAdcTtyReader | None = None try: tty.setraw(slave_fd) reader = KamilAdcTtyReader(os.ttyname(slave_fd)) reader.open() os.write( master_fd, _start_frame() + _point_frame(1, 10, -1) + _point_frame(2, -20, 2) + _start_frame() + _point_frame(1, 30, -3) + _start_frame(), ) first = reader.read_sweep(timeout_s=1.0) second = reader.read_sweep(timeout_s=1.0) self.assertEqual(first.tolist(), [complex(10, -1), complex(-20, 2)]) self.assertEqual(second.tolist(), [complex(30, -3)]) finally: if reader is not None: reader.close() os.close(master_fd) os.close(slave_fd) def test_expected_point_count_discards_mismatched_sweep(self) -> None: master_fd, slave_fd = pty.openpty() reader: KamilAdcTtyReader | None = None try: tty.setraw(slave_fd) reader = KamilAdcTtyReader(os.ttyname(slave_fd)) reader.open() os.write( master_fd, _start_frame() + _point_frame(1, 5, -5) + _start_frame() + _point_frame(1, 10, -1) + _point_frame(2, -20, 2) + _start_frame(), ) values = reader.read_sweep(timeout_s=1.0, expected_points=2) self.assertEqual(values.tolist(), [complex(10, -1), complex(-20, 2)]) finally: if reader is not None: reader.close() os.close(master_fd) os.close(slave_fd) def test_stream_without_next_start_times_out_with_received_count(self) -> None: master_fd, slave_fd = pty.openpty() reader: KamilAdcTtyReader | None = None try: tty.setraw(slave_fd) reader = KamilAdcTtyReader(os.ttyname(slave_fd)) reader.open() os.write(master_fd, _start_frame() + _point_frame(1, 10, -1)) with self.assertRaisesRegex(TimeoutError, "sweep end: received 1 points"): reader.read_sweep(timeout_s=0.05) finally: if reader is not None: reader.close() os.close(master_fd) os.close(slave_fd) class KamilAdcConfigTest(unittest.TestCase): def test_config_round_trip_preserves_kamil_sections(self) -> None: payload = { "radar": { "model": "kamil_adc", "serial": "kamil_adc", "driver_mode": "native", "kamil_adc": { "project_dir": "/home/europa/Documents/kamil_adc", "executable_path": "/home/europa/Documents/kamil_adc/kamil_adc_capture", "tty_path": "/tmp/ttyADC_data", "args": ["profile:phase", "do1_pair_subtract_avg"], "env": {"ADC_ENV": "1"}, "startup_timeout_s": 7.0, "sweep_timeout_s": 8.0, "stop_timeout_s": 3.0, }, "laser_control": { "enabled": True, "port": "/dev/ttyUSB0", "mode": "variation", "pi_coeff1_p": 2560, "pi_coeff1_i": 128, "pi_coeff2_p": 2600, "pi_coeff2_i": 140, "manual": { "temp1": 26.0, "temp2": 27.0, "current1": 31.0, "current2": 32.0, }, "variation": { "variation_type": "CHANGE_CURRENT_LD2", "static_temp1": 28.0, "static_temp2": 29.0, "static_current1": 33.0, "static_current2": 34.0, "min_value": 30.0, "max_value": 40.0, "step": 0.5, "time_step": 50, "delay_time": 10, }, }, "sweep": { "start_hz": 1.0, "stop_hz": 2.0, "points": 2, "if_bandwidth_hz": 1.0, "stimulus_power_dbm": -10.0, }, }, "switches": { "port1": {"positions": 1}, "port2": {"positions": 1}, }, } encoded = RunConfigModel.from_dict(payload).to_dict() self.assertEqual(encoded["radar"]["model"], "kamil_adc") self.assertNotIn("points", encoded["radar"]["sweep"]) self.assertEqual(encoded["radar"]["kamil_adc"]["tty_path"], "/tmp/ttyADC_data") self.assertEqual(encoded["radar"]["kamil_adc"]["args"], ["profile:phase", "do1_pair_subtract_avg"]) self.assertEqual(encoded["radar"]["kamil_adc"]["env"], {"ADC_ENV": "1"}) self.assertEqual(encoded["radar"]["laser_control"]["mode"], "variation") self.assertEqual( encoded["radar"]["laser_control"]["variation"]["variation_type"], "CHANGE_CURRENT_LD2", ) def test_supervisor_selects_kamil_adc_producer(self) -> None: with tempfile.TemporaryDirectory() as tmp_dir: config_path = Path(tmp_dir) / "run_config.json" config_path.write_text(json.dumps({"radar": {"model": "kamil_adc"}}), encoding="utf-8") command = ProcessSupervisor(Path("/repo"))._acquisition_command(config_path) self.assertEqual( command, [ sys.executable, "-m", "python_app.scripts.kamil_adc_raw_producer", "--config", str(config_path), ], ) if __name__ == "__main__": unittest.main()