224 lines
8.9 KiB
Python
224 lines
8.9 KiB
Python
"""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()
|