277 lines
10 KiB
Python
277 lines
10 KiB
Python
"""Tests for Kamil ADC config, frame parsing, TTY reader, and producer wiring."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import pty
|
|
import struct
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import tty
|
|
import unittest
|
|
|
|
from python_app.hardware_full.kamil_adc_service import (
|
|
KamilAdcTtyReader,
|
|
_parse_point_frame,
|
|
)
|
|
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 ParsePointFrameTest(unittest.TestCase):
|
|
def test_parses_valid_point(self) -> None:
|
|
value = _parse_point_frame(_point_frame(1, 123, -45), expected_step=1)
|
|
self.assertEqual(value, complex(123, -45))
|
|
|
|
def test_rejects_bad_marker(self) -> None:
|
|
with self.assertRaisesRegex(ValueError, "marker mismatch"):
|
|
_parse_point_frame(_point_frame(1, 10, 20, marker=0x001A), expected_step=1)
|
|
|
|
def test_rejects_wrong_step(self) -> None:
|
|
with self.assertRaisesRegex(ValueError, "step mismatch"):
|
|
_parse_point_frame(_point_frame(2, 10, 20), expected_step=1)
|
|
|
|
|
|
class KamilAdcTtyReaderTest(unittest.TestCase):
|
|
"""End-to-end tests over a PTY exercising the background reader thread."""
|
|
|
|
def _open_pty_reader(self) -> tuple[int, int, KamilAdcTtyReader]:
|
|
master_fd, slave_fd = pty.openpty()
|
|
tty.setraw(slave_fd)
|
|
reader = KamilAdcTtyReader(os.ttyname(slave_fd))
|
|
reader.open()
|
|
return master_fd, slave_fd, reader
|
|
|
|
@staticmethod
|
|
def _close(master_fd: int, slave_fd: int, reader: KamilAdcTtyReader) -> None:
|
|
try:
|
|
reader.close()
|
|
finally:
|
|
os.close(master_fd)
|
|
os.close(slave_fd)
|
|
|
|
def test_publishes_first_complete_sweep(self) -> None:
|
|
master_fd, slave_fd, reader = self._open_pty_reader()
|
|
try:
|
|
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)])
|
|
self.assertEqual(reader.locked_points, 2)
|
|
finally:
|
|
self._close(master_fd, slave_fd, reader)
|
|
|
|
def test_consecutive_constant_length_sweeps(self) -> None:
|
|
"""Each newly-completed sweep is delivered once new data arrives after a read."""
|
|
master_fd, slave_fd, reader = self._open_pty_reader()
|
|
try:
|
|
os.write(
|
|
master_fd,
|
|
_start_frame()
|
|
+ _point_frame(1, 10, -1)
|
|
+ _point_frame(2, -20, 2)
|
|
+ _start_frame(),
|
|
)
|
|
first = reader.read_sweep(timeout_s=1.0)
|
|
self.assertEqual(first.tolist(), [complex(10, -1), complex(-20, 2)])
|
|
|
|
os.write(
|
|
master_fd,
|
|
_point_frame(1, 30, -3) + _point_frame(2, -40, 4) + _start_frame(),
|
|
)
|
|
second = reader.read_sweep(timeout_s=1.0)
|
|
self.assertEqual(second.tolist(), [complex(30, -3), complex(-40, 4)])
|
|
finally:
|
|
self._close(master_fd, slave_fd, reader)
|
|
|
|
def test_shorter_sweep_after_lock_raises(self) -> None:
|
|
"""A later sweep with fewer points than the locked-in count fails fast."""
|
|
master_fd, slave_fd, reader = self._open_pty_reader()
|
|
try:
|
|
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)
|
|
self.assertEqual(first.tolist(), [complex(10, -1), complex(-20, 2)])
|
|
with self.assertRaisesRegex(RuntimeError, "sweep length changed"):
|
|
reader.read_sweep(timeout_s=1.0)
|
|
finally:
|
|
self._close(master_fd, slave_fd, reader)
|
|
|
|
def test_longer_sweep_after_lock_raises(self) -> None:
|
|
"""A later sweep with more points than the locked-in count fails fast."""
|
|
master_fd, slave_fd, reader = self._open_pty_reader()
|
|
try:
|
|
os.write(
|
|
master_fd,
|
|
_start_frame()
|
|
+ _point_frame(1, 10, -1)
|
|
+ _start_frame()
|
|
+ _point_frame(1, 30, -3)
|
|
+ _point_frame(2, -40, 4)
|
|
+ _start_frame(),
|
|
)
|
|
first = reader.read_sweep(timeout_s=1.0)
|
|
self.assertEqual(first.tolist(), [complex(10, -1)])
|
|
with self.assertRaisesRegex(RuntimeError, "exceeded locked point count"):
|
|
reader.read_sweep(timeout_s=1.0)
|
|
finally:
|
|
self._close(master_fd, slave_fd, reader)
|
|
|
|
def test_no_completed_sweep_times_out(self) -> None:
|
|
master_fd, slave_fd, reader = self._open_pty_reader()
|
|
try:
|
|
# Start marker plus a partial sweep with no follow-up boundary.
|
|
os.write(master_fd, _start_frame() + _point_frame(1, 10, -1))
|
|
with self.assertRaisesRegex(TimeoutError, "Timed out waiting for Kamil ADC sweep"):
|
|
reader.read_sweep(timeout_s=0.1)
|
|
finally:
|
|
self._close(master_fd, slave_fd, reader)
|
|
|
|
def test_only_latest_sweep_is_published(self) -> None:
|
|
"""If multiple sweeps arrive before the consumer reads, only the newest survives."""
|
|
master_fd, slave_fd, reader = self._open_pty_reader()
|
|
try:
|
|
payload = (
|
|
_start_frame()
|
|
+ _point_frame(1, 1, 0)
|
|
+ _point_frame(2, 2, 0)
|
|
+ _start_frame()
|
|
+ _point_frame(1, 3, 0)
|
|
+ _point_frame(2, 4, 0)
|
|
+ _start_frame()
|
|
+ _point_frame(1, 5, 0)
|
|
+ _point_frame(2, 6, 0)
|
|
+ _start_frame()
|
|
)
|
|
os.write(master_fd, payload)
|
|
# Wait until the reader thread has parsed all three sweeps before
|
|
# reading from the mailbox — otherwise we'd race the producer and
|
|
# might consume an intermediate value.
|
|
deadline = time.monotonic() + 1.0
|
|
while time.monotonic() < deadline and reader.published_count < 3:
|
|
time.sleep(0.005)
|
|
self.assertGreaterEqual(reader.published_count, 3)
|
|
values = reader.read_sweep(timeout_s=1.0)
|
|
# The reader thread overwrites unread sweeps; the consumer sees the
|
|
# most recently completed one.
|
|
self.assertEqual(values.tolist(), [complex(5, 0), complex(6, 0)])
|
|
finally:
|
|
self._close(master_fd, slave_fd, reader)
|
|
|
|
|
|
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()
|