170 lines
6.2 KiB
Python
170 lines
6.2 KiB
Python
"""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("<HHHH", 0x000A, 0xFFFF, 0xFFFF, 0xFFFF)
|
|
|
|
|
|
def _point_frame(step: int, real: int, imag: int, *, marker: int = 0x000A) -> bytes:
|
|
return struct.pack("<HHhh", marker, step, real, imag)
|
|
|
|
|
|
class KamilAdcFrameParserTest(unittest.TestCase):
|
|
def test_parse_valid_point(self) -> 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))
|
|
|
|
values = reader.read_sweep(points=2, 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_short_stream_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, "received 1/2"):
|
|
reader.read_sweep(points=2, 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.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()
|