"""Tests for the laser current-variation temperature monitoring package.""" from __future__ import annotations import tempfile import threading import unittest from dataclasses import dataclass from pathlib import Path from typing import List, Optional from python_app.hardware_full.laser_control.monitoring import ( LaserTemperatureChecker, LaserTemperatureMonitor, LaserVariationSession, ReadingReader, ReadingWriter, TemperatureReading, compute_sweep_period_s, resolve_period_s, ) @dataclass class _FakeMeasurements: temp1: float temp2: float temp_ext1: Optional[float] = None temp_ext2: Optional[float] = None current1: Optional[float] = None current2: Optional[float] = None class _FakeController: """Returns a queued sequence of measurements, then None.""" def __init__(self, measurements: List[Optional[_FakeMeasurements]]) -> None: self._queue = list(measurements) def get_measurements(self) -> Optional[_FakeMeasurements]: return self._queue.pop(0) if self._queue else None class SessionRoundTripTest(unittest.TestCase): def test_save_then_load_preserves_targets_and_tolerance(self) -> None: with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "session.json" LaserVariationSession( variation_type="CHANGE_CURRENT_LD1", target_temp1=28.0, target_temp2=28.9, tolerance_c=0.03, started_at_iso="2026-07-27T12:00:00", ).save(path) loaded = LaserVariationSession.load(path) self.assertEqual(loaded.variation_type, "CHANGE_CURRENT_LD1") self.assertAlmostEqual(loaded.target_temp1, 28.0) self.assertAlmostEqual(loaded.target_temp2, 28.9) self.assertAlmostEqual(loaded.tolerance_c, 0.03) def test_load_missing_file_raises(self) -> None: with tempfile.TemporaryDirectory() as tmp: with self.assertRaises(FileNotFoundError): LaserVariationSession.load(Path(tmp) / "absent.json") class ReadingsChannelTest(unittest.TestCase): def _reading(self, seq: int, t1: float = 25.0, t2: float = 25.0) -> TemperatureReading: return TemperatureReading(seq=seq, mono_ns=seq, temp1=t1, temp2=t2) def test_reader_tails_appended_lines_in_order(self) -> None: with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "readings.jsonl" reader = ReadingReader(path) # start at (nonexistent) end with ReadingWriter(path) as writer: writer.write(self._reading(0, 25.0)) writer.write(self._reading(1, 26.0)) first = list(reader.poll()) writer.write(self._reading(2, 27.0)) second = list(reader.poll()) self.assertEqual([r.seq for r in first], [0, 1]) self.assertEqual([r.seq for r in second], [2]) self.assertAlmostEqual(first[1].temp1, 26.0) def test_partial_trailing_line_is_buffered_until_newline(self) -> None: with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "readings.jsonl" path.write_text('{"seq":0,"mono_ns":0,"temp1":25.0,"temp2":25.0}\n{"seq":1,"mono', encoding="utf-8") reader = ReadingReader(path, from_start=True) first = list(reader.poll()) # Complete the truncated line. with path.open("a", encoding="utf-8") as fh: fh.write('_ns":1,"temp1":26.0,"temp2":26.0}\n') second = list(reader.poll()) self.assertEqual([r.seq for r in first], [0]) self.assertEqual([r.seq for r in second], [1]) class SweepPeriodTest(unittest.TestCase): def test_compute_sweep_period_matches_formula(self) -> None: # (35-33)/0.05 = 40 -> 41 points; per point = 10ms + 50us = 0.01005s. period = compute_sweep_period_s(33.0, 35.0, 0.05, time_step_us=50, delay_time_ms=10) self.assertAlmostEqual(period, 41 * 0.01005, places=6) def test_resolve_interval_strategy(self) -> None: period = resolve_period_s( "interval:250", min_value=33.0, max_value=35.0, step=0.05, time_step_us=50, delay_time_ms=10, ) self.assertAlmostEqual(period, 0.25) def test_resolve_rejects_unknown_strategy(self) -> None: with self.assertRaises(ValueError): resolve_period_s("bogus", min_value=0, max_value=1, step=0.1, time_step_us=50, delay_time_ms=10) class MonitorTest(unittest.TestCase): def test_read_once_maps_measurement_fields(self) -> None: controller = _FakeController([_FakeMeasurements( temp1=28.01, temp2=28.9, temp_ext1=22.0, temp_ext2=23.0, current1=33.0, current2=35.0, )]) with tempfile.TemporaryDirectory() as tmp: with ReadingWriter(Path(tmp) / "r.jsonl") as writer: monitor = LaserTemperatureMonitor(controller, writer, period_s=0.0) reading = monitor.read_once(7) assert reading is not None self.assertEqual(reading.seq, 7) self.assertAlmostEqual(reading.temp1, 28.01) self.assertAlmostEqual(reading.temp_ext1, 22.0) self.assertAlmostEqual(reading.current2, 35.0) def test_run_publishes_until_stopped(self) -> None: controller = _FakeController([ _FakeMeasurements(28.0, 28.9), _FakeMeasurements(28.0, 28.9), ]) stop = threading.Event() class _OneShotWriter: def __init__(self) -> None: self.written: List[TemperatureReading] = [] def write(self, reading: TemperatureReading) -> None: self.written.append(reading) stop.set() # stop after the first publish writer = _OneShotWriter() monitor = LaserTemperatureMonitor(controller, writer, period_s=0.0) monitor.run(stop) self.assertEqual(len(writer.written), 1) self.assertEqual(writer.written[0].seq, 0) class CheckerTest(unittest.TestCase): def _checker(self, **kwargs: object) -> LaserTemperatureChecker: return LaserTemperatureChecker(target_temp1=28.0, target_temp2=28.9, tolerance_c=0.03, **kwargs) def _reading(self, t1: float, t2: float, seq: int = 0) -> TemperatureReading: return TemperatureReading(seq=seq, mono_ns=seq, temp1=t1, temp2=t2) def test_laser1_off_target_warns_once_for_laser1(self) -> None: checker = self._checker() warned = checker.process(self._reading(t1=28.05, t2=28.9)) # laser1 off by 0.05 self.assertEqual([d.laser for d in warned], [1]) def test_within_tolerance_no_warning(self) -> None: checker = self._checker() warned = checker.process(self._reading(t1=28.01, t2=28.9)) # 0.01 < 0.03 self.assertEqual(warned, []) def test_boundary_equal_tolerance_is_ok(self) -> None: checker = self._checker() warned = checker.process(self._reading(t1=28.03, t2=28.9)) # |Δ|==tol -> within self.assertEqual(warned, []) def test_both_lasers_off_target_warn_independently(self) -> None: checker = self._checker() warned = checker.process(self._reading(t1=27.9, t2=29.0)) self.assertEqual(sorted(d.laser for d in warned), [1, 2]) def test_persistent_mismatch_warns_once_then_silent(self) -> None: checker = self._checker() first = checker.process(self._reading(t1=28.1, t2=28.9, seq=0)) second = checker.process(self._reading(t1=28.1, t2=28.9, seq=1)) self.assertEqual([d.laser for d in first], [1]) self.assertEqual(second, []) # no reminder configured def test_reminder_repeats_warning(self) -> None: checker = self._checker(reminder_every=2) checker.process(self._reading(t1=28.1, t2=28.9, seq=0)) # initial warn self.assertEqual(checker.process(self._reading(t1=28.1, t2=28.9, seq=1)), []) again = checker.process(self._reading(t1=28.1, t2=28.9, seq=2)) # reminder self.assertEqual([d.laser for d in again], [1]) def test_recovery_clears_mismatch_state(self) -> None: checker = self._checker() checker.process(self._reading(t1=28.1, t2=28.9, seq=0)) # warn checker.process(self._reading(t1=28.0, t2=28.9, seq=1)) # recover (info, no warn) rewarn = checker.process(self._reading(t1=28.1, t2=28.9, seq=2)) # warns again self.assertEqual([d.laser for d in rewarn], [1]) def test_from_session_uses_session_targets(self) -> None: session = LaserVariationSession( variation_type="CHANGE_CURRENT_LD1", target_temp1=30.0, target_temp2=31.0, tolerance_c=0.03, ) checker = LaserTemperatureChecker.from_session(session) warned = checker.process(self._reading(t1=30.1, t2=31.0)) self.assertEqual([d.laser for d in warned], [1]) if __name__ == "__main__": unittest.main()