diff --git a/python_app/hardware_full/kamil_adc/laser.py b/python_app/hardware_full/kamil_adc/laser.py index 3ee22d5..9fcaf17 100644 --- a/python_app/hardware_full/kamil_adc/laser.py +++ b/python_app/hardware_full/kamil_adc/laser.py @@ -86,12 +86,38 @@ def apply_kamil_adc_laser_control(config: RunConfigModel) -> bool: "delay_time": variation.delay_time, }, ) + _write_variation_session(variation) return True raise RuntimeError(f"Unsupported laser_control mode: {laser.mode}") finally: controller.disconnect() +def _write_variation_session(variation) -> None: + """Freeze the variation's static temperature targets for the checker. + + Best-effort: a failure to write the session snapshot must never abort the + acquisition setup, so any error is logged and swallowed. + """ + from datetime import datetime + + try: + from python_app.hardware_full.laser_control.monitoring.session import ( + LaserVariationSession, + ) + + LaserVariationSession( + variation_type=variation.variation_type, + target_temp1=variation.static_temp1, + target_temp2=variation.static_temp2, + tolerance_c=variation.temp_tolerance_c, + started_at_iso=datetime.now().isoformat(timespec="seconds"), + ).save() + logger.debug("Wrote laser variation session snapshot for the temperature checker") + except Exception: # noqa: BLE001 — session snapshot is auxiliary, never fatal + logger.warning("Failed to write laser variation session snapshot", exc_info=True) + + def _validate_laser_control_config(config: RunConfigModel) -> None: laser = config.radar.laser_control if not laser.port: diff --git a/python_app/hardware_full/laser_control/monitoring/README.md b/python_app/hardware_full/laser_control/monitoring/README.md new file mode 100644 index 0000000..38362f9 --- /dev/null +++ b/python_app/hardware_full/laser_control/monitoring/README.md @@ -0,0 +1,247 @@ +# Контроль температуры при вариации тока лазера + +Набор из трёх развязанных компонентов для автоматизации измерений в режиме +**вариации тока лазера 1** (`CHANGE_CURRENT_LD1`). Пока плата гоняет свип тока, +температуры лазеров должны оставаться на заданных статичных уставках. Эти модули +раз в свип считывают реальную температуру и предупреждают, если она разошлась с +целью. + +## Зачем это нужно + +При запуске вариации тока из GUI изменённые значения температуры могут фактически +не дойти до цели — реальная температура остаётся прежней, и измерение становится +некорректным. Плата после старта задачи гоняет свип **автономно** и никак не +сигнализирует, что уставка не достигнута. Эти модули закрывают пробел: независимо +опрашивают плату и валидируют температуру относительно уставок, зафиксированных +**в момент старта вариации**. + +> ⚠️ В прошивке реализована только **вариация тока** (`CHANGE_CURRENT_LD1`). +> Вариация температуры не поддерживается и в этот API не заложена. + +## Архитектура + +``` +[starter] ── TASK_ENABLE ──► плата кратко открыл порт, послал, закрыл + │ пишет session.json (target temp1/2, tolerance, variation_type) + ▼ +[monitor] ── TRANS_ENABLE ──► плата владеет портом всё время работы + │ раз в свип: get_measurements() + │ дописывает строку в readings.jsonl (seq, temp1, temp2, temp_ext, I1, I2) + ▼ +[checker] читает session.json + tail readings.jsonl + сверяет temp1↔target_temp1 и temp2↔target_temp2, |Δ|>tol ─► WARNING в консоль +``` + +- **Порт лазера эксклюзивен.** `starter` трогает его кратко, затем `monitor` + владеет им всё время. `checker` порт не трогает вовсе — читает только файлы. +- **Связь через файлы** (JSONL + JSON), а не сокеты, — процессы стартуют, + останавливаются и перезапускаются независимо, без рукопожатия. +- **Сверяются оба лазера** по внутренним `temp1`/`temp2` (не по внешним + термисторам `temp_ext*`), каждый со своим допуском (по умолчанию `0.03 °C`). + +## Быстрый старт (CLI, два процесса) + +Терминал 1 — стартовать вариацию и мониторить температуру: + +```bash +python -m python_app.scripts.laser_temp_monitor \ + --config run_config.json \ + --start +``` + +Терминал 2 — валидировать температуру и печатать предупреждения: + +```bash +python -m python_app.scripts.laser_temp_checker +``` + +Пример вывода чекера при расхождении и возврате в допуск: + +``` +WARNING laser_temp_checker: Laser 1 temperature off target: measured 28.100 °C, + target 28.000 °C, Δ=+0.100 °C exceeds tolerance ±0.030 °C [seq=1] +INFO laser_temp_checker: Laser 1 temperature back within tolerance: + 28.000 °C (target 28.000, |Δ|=0.000 ≤ 0.030) [seq=2] +``` + +Остановка — `Ctrl+C` (SIGINT) в любом из процессов. + +## Конфигурация + +Параметры берутся из `run_config.json`, секция `radar.laser_control`. Мониторинг +использует блок `variation` и новое поле `temp_tolerance_c`: + +```json +{ + "radar": { + "model": "kamil_adc", + "laser_control": { + "enabled": true, + "port": "/dev/ttyUSB0", + "mode": "variation", + "pi_coeff1_p": 2560, + "pi_coeff1_i": 128, + "pi_coeff2_p": 2560, + "pi_coeff2_i": 128, + "variation": { + "variation_type": "CHANGE_CURRENT_LD1", + "static_temp1": 28.0, + "static_temp2": 28.9, + "static_current1": 33.0, + "static_current2": 35.0, + "min_value": 33.0, + "max_value": 60.0, + "step": 0.05, + "time_step": 50, + "delay_time": 10, + "temp_tolerance_c": 0.03 + } + } + } +} +``` + +Ключевые поля для мониторинга: + +| Поле | Смысл | +|---|---| +| `port` | Серийный порт лазерной платы (пусто → автоопределение) | +| `static_temp1` / `static_temp2` | Целевые статичные температуры лазеров 1/2, °C | +| `min_value` / `max_value` / `step` | Диапазон и шаг свипа тока, мА — из них считается период свипа | +| `time_step` / `delay_time` | Тайминги точки (мкс / мс) — тоже входят в период свипа | +| `temp_tolerance_c` | Допуск сверки, °C (по умолчанию `0.03`) | + +## Опции CLI + +### `laser_temp_monitor` + +| Аргумент | По умолчанию | Назначение | +|---|---|---| +| `--config` | — (обязателен) | Путь к `run_config.json` | +| `--start` | выкл. | Послать `CHANGE_CURRENT_LD1` перед мониторингом и записать сессию | +| `--strategy` | `computed` | `computed` (раз в свип) или `interval:` (фикс. период) | +| `--readings` | `/laser_temp_readings.jsonl` | Куда дописывать показания | +| `--session` | `/laser_variation_session.json` | Куда писать снимок сессии (с `--start`) | + +Мониторить уже запущенную из GUI/пайплайна вариацию (без повторного старта): + +```bash +python -m python_app.scripts.laser_temp_monitor --config run_config.json +``` + +Фиксированный период вместо расчётного (напр. раз в 500 мс): + +```bash +python -m python_app.scripts.laser_temp_monitor \ + --config run_config.json --strategy interval:500 +``` + +### `laser_temp_checker` + +| Аргумент | По умолчанию | Назначение | +|---|---|---| +| `--session` | `/laser_variation_session.json` | Снимок с целями и допуском | +| `--readings` | `/laser_temp_readings.jsonl` | Какой канал показаний тайлить | +| `--tolerance` | из сессии | Переопределить допуск, °C | +| `--reminder-every` | `0` (выкл.) | Повторять предупреждение каждые N показаний, пока вне допуска | +| `--from-start` | выкл. | Проверить весь файл показаний, а не только новые строки | + +Разные пути для нескольких одновременных прогонов: + +```bash +# монитор +python -m python_app.scripts.laser_temp_monitor --config cfg.json --start \ + --readings /tmp/run7.jsonl --session /tmp/run7.session.json +# чекер +python -m python_app.scripts.laser_temp_checker \ + --readings /tmp/run7.jsonl --session /tmp/run7.session.json --reminder-every 20 +``` + +## Интеграция с пайплайном Kamil ADC + +Когда вариацию стартует штатный пайплайн +([`apply_kamil_adc_laser_control`](../../kamil_adc/laser.py)), снимок сессии +`session.json` пишется автоматически. Достаточно запустить только чекер +(и, при желании, монитор без `--start`, чтобы он опрашивал плату). Так консоль +получит предупреждения о рассинхроне температуры прямо во время захвата. + +## Встраивание в свой код (без CLI) + +```python +import threading +from python_app.hardware_full.laser_control.controller import LaserController +from python_app.hardware_full.laser_control.monitoring import ( + LaserTemperatureMonitor, LaserTemperatureChecker, LaserVariationSession, + ReadingWriter, ReadingReader, resolve_period_s, +) + +# 1. Зафиксировать цели при старте вариации +session = LaserVariationSession( + variation_type="CHANGE_CURRENT_LD1", + target_temp1=28.0, target_temp2=28.9, tolerance_c=0.03, +) + +# 2. Монитор (в проде controller — реальный LaserController) +period = resolve_period_s("computed", min_value=33.0, max_value=60.0, step=0.05, + time_step_us=50, delay_time_ms=10) +stop = threading.Event() +with LaserController(port="/dev/ttyUSB0") as ctrl, ReadingWriter("readings.jsonl") as w: + monitor = LaserTemperatureMonitor(ctrl, w, period_s=period) + threading.Thread(target=monitor.run, args=(stop,), daemon=True).start() + + # 3. Чекер: тайлить показания и валидировать оба лазера + checker = LaserTemperatureChecker.from_session(session) + reader = ReadingReader("readings.jsonl") + while not stop.is_set(): + for reading in reader.poll(): + checker.process(reading) # печатает WARNING при |Δ| > tolerance + stop.wait(0.2) +``` + +`LaserTemperatureChecker.evaluate(reading)` возвращает список +`LaserDeviation` (по лазеру: измеренное, цель, Δ, в допуске ли) без логирования — +удобно для собственной обработки/накопления статистики. + +## Формат IPC-файлов + +`session.json`: + +```json +{ + "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" +} +``` + +`readings.jsonl` (по одной строке-объекту на свип): + +```json +{"seq":0,"mono_ns":123456789,"temp1":28.0,"temp2":28.9,"temp_ext1":22.0,"temp_ext2":23.0,"current1":33.0,"current2":35.0} +``` + +## Как определяется «раз в свип» + +Плата не отдаёт явную границу свипа, поэтому период оценивается из параметров: + +``` +num_steps = round(|max_value - min_value| / step) + 1 +per_point_s = delay_time / 1000 + time_step / 1_000_000 +sweep_period = num_steps × per_point_s +``` + +Монитор публикует одно показание за такой период. Если нужен другой темп — +`--strategy interval:`. (Внутренний счётчик `TO6` платы существует, но его +семантика не гарантирована, поэтому для тайминга он не используется.) + +## Тесты + +```bash +python -m pytest python_app/tests/test_laser_temp_monitoring.py -q +``` + +Покрыто: round-trip сессии, tail JSONL (включая усечённую последнюю строку), +расчёт периода свипа, маппинг измерений монитором, и валидация чекера по каждому +лазеру отдельно (порог, граница допуска, повторные предупреждения, восстановление). diff --git a/python_app/hardware_full/laser_control/monitoring/__init__.py b/python_app/hardware_full/laser_control/monitoring/__init__.py new file mode 100644 index 0000000..e241424 --- /dev/null +++ b/python_app/hardware_full/laser_control/monitoring/__init__.py @@ -0,0 +1,36 @@ +"""Laser current-variation temperature monitoring and validation. + +Three decoupled pieces connected via IPC files: + - :class:`LaserVariationSession` — target setpoints + tolerance frozen at start. + - :class:`LaserTemperatureMonitor` — polls the board once per sweep, publishes. + - :class:`LaserTemperatureChecker` — validates published readings, warns. +""" + +from .checker import LaserDeviation, LaserTemperatureChecker +from .monitor import ( + LaserTemperatureMonitor, + compute_sweep_period_s, + resolve_period_s, +) +from .readings_channel import ReadingReader, ReadingWriter, TemperatureReading +from .session import ( + DEFAULT_READINGS_PATH, + DEFAULT_SESSION_PATH, + DEFAULT_TOLERANCE_C, + LaserVariationSession, +) + +__all__ = [ + "LaserDeviation", + "LaserTemperatureChecker", + "LaserTemperatureMonitor", + "compute_sweep_period_s", + "resolve_period_s", + "ReadingReader", + "ReadingWriter", + "TemperatureReading", + "DEFAULT_READINGS_PATH", + "DEFAULT_SESSION_PATH", + "DEFAULT_TOLERANCE_C", + "LaserVariationSession", +] diff --git a/python_app/hardware_full/laser_control/monitoring/checker.py b/python_app/hardware_full/laser_control/monitoring/checker.py new file mode 100644 index 0000000..103624b --- /dev/null +++ b/python_app/hardware_full/laser_control/monitoring/checker.py @@ -0,0 +1,139 @@ +"""Independent laser temperature checker. + +Reads the target setpoints frozen at variation start (:class:`LaserVariationSession`) +and validates each published :class:`TemperatureReading` against them. Both lasers +are checked independently: ``temp1`` against ``target_temp1`` and ``temp2`` against +``target_temp2``. When a laser's measured temperature deviates from its target by +more than the tolerance (default 0.03 °C), a warning is printed to the console. + +Runs as its own process (see ``scripts/laser_temp_checker.py``), reading the JSONL +readings channel — it never touches the serial port, so it is fully independent of +the monitor and can be started, stopped, or restarted at any time. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import List + +from .readings_channel import TemperatureReading +from .session import DEFAULT_TOLERANCE_C, LaserVariationSession + +logger = logging.getLogger(__name__) + +# Only a deviation strictly greater than the tolerance warns; this epsilon keeps a +# value the user intends to be exactly at the tolerance from tripping on float error +# (e.g. 28.03 - 28.00 == 0.030000000000001 in IEEE-754). +_FLOAT_EPS = 1e-9 + + +@dataclass(slots=True) +class LaserDeviation: + """Result of comparing one laser's measured temperature to its target.""" + + laser: int # 1 or 2 + seq: int + measured: float + target: float + delta: float # measured - target, °C + within_tolerance: bool + + +class LaserTemperatureChecker: + """Validates readings against per-laser targets and warns on mismatch. + + Anti-spam: a laser's ok↔mismatch transitions are logged once; while a laser + stays out of tolerance, a reminder is emitted only every ``reminder_every`` + readings (0 disables reminders). State is tracked independently per laser, so + a persistent laser-1 fault never suppresses a fresh laser-2 warning. + """ + + def __init__( + self, + target_temp1: float, + target_temp2: float, + tolerance_c: float = DEFAULT_TOLERANCE_C, + reminder_every: int = 0, + ) -> None: + self.target_temp1 = float(target_temp1) + self.target_temp2 = float(target_temp2) + self.tolerance_c = float(tolerance_c) + self.reminder_every = int(reminder_every) + # Per-laser state: mismatch flag + readings seen since the last log. + self._mismatch = {1: False, 2: False} + self._since_log = {1: 0, 2: 0} + + @classmethod + def from_session( + cls, session: LaserVariationSession, reminder_every: int = 0 + ) -> "LaserTemperatureChecker": + return cls( + target_temp1=session.target_temp1, + target_temp2=session.target_temp2, + tolerance_c=session.tolerance_c, + reminder_every=reminder_every, + ) + + def evaluate(self, reading: TemperatureReading) -> List[LaserDeviation]: + """Compute per-laser deviations without logging (pure).""" + return [ + self._deviation(1, reading.seq, reading.temp1, self.target_temp1), + self._deviation(2, reading.seq, reading.temp2, self.target_temp2), + ] + + def process(self, reading: TemperatureReading) -> List[LaserDeviation]: + """Evaluate a reading and emit console warnings, honouring anti-spam. + + Returns the deviations for which a warning/reminder was emitted this call + (empty when both lasers are within tolerance and unchanged). + """ + warned: List[LaserDeviation] = [] + for dev in self.evaluate(reading): + if self._should_warn(dev): + self._warn(dev) + warned.append(dev) + return warned + + def _deviation(self, laser: int, seq: int, measured: float, target: float) -> LaserDeviation: + delta = measured - target + return LaserDeviation( + laser=laser, + seq=seq, + measured=measured, + target=target, + delta=delta, + within_tolerance=abs(delta) <= self.tolerance_c + _FLOAT_EPS, + ) + + def _should_warn(self, dev: LaserDeviation) -> bool: + laser = dev.laser + if not dev.within_tolerance: + if not self._mismatch[laser]: + # Fresh ok -> mismatch transition: always warn. + self._mismatch[laser] = True + self._since_log[laser] = 0 + return True + # Still out of tolerance: warn again only every reminder_every readings. + self._since_log[laser] += 1 + if self.reminder_every > 0 and self._since_log[laser] >= self.reminder_every: + self._since_log[laser] = 0 + return True + return False + # Within tolerance: log a recovery once, then stay quiet. + if self._mismatch[laser]: + self._mismatch[laser] = False + self._since_log[laser] = 0 + logger.info( + "Laser %d temperature back within tolerance: %.3f °C " + "(target %.3f, |Δ|=%.3f ≤ %.3f) [seq=%d]", + laser, dev.measured, dev.target, abs(dev.delta), self.tolerance_c, dev.seq, + ) + return False + + def _warn(self, dev: LaserDeviation) -> None: + logger.warning( + "Laser %d temperature off target: measured %.3f °C, target %.3f °C, " + "Δ=%+.3f °C exceeds tolerance ±%.3f °C [seq=%d]", + dev.laser, dev.measured, dev.target, dev.delta, self.tolerance_c, dev.seq, + ) diff --git a/python_app/hardware_full/laser_control/monitoring/monitor.py b/python_app/hardware_full/laser_control/monitoring/monitor.py new file mode 100644 index 0000000..e44e5df --- /dev/null +++ b/python_app/hardware_full/laser_control/monitoring/monitor.py @@ -0,0 +1,142 @@ +"""Independent laser temperature monitor. + +Owns a :class:`LaserController` connection and, once per current-variation sweep, +polls the board for a measurement and publishes it to a JSONL readings channel. +The board runs the current sweep autonomously after ``TASK_ENABLE``; the monitor +only reads the "last data point" via ``TRANS_ENABLE`` — exactly like the original +RadioPhotonic PC software's polling loop, but decoupled and headless. + +Runs as its own process (see ``scripts/laser_temp_monitor.py``) so it is fully +independent of both the acquisition pipeline and the temperature checker. +""" + +from __future__ import annotations + +import logging +import threading +import time +from dataclasses import dataclass +from typing import Optional, Protocol + +from .readings_channel import ReadingWriter, TemperatureReading + +logger = logging.getLogger(__name__) + + +class _MeasurementSource(Protocol): + """Minimal controller surface the monitor depends on (eases testing).""" + + def get_measurements(self) -> object: ... + + +def compute_sweep_period_s( + min_value: float, + max_value: float, + step: float, + time_step_us: float, + delay_time_ms: float, +) -> float: + """Estimate the duration of one min→max current sweep, in seconds. + + ``num_steps = round(|max - min| / step) + 1`` points, each taking roughly the + inter-point delay plus the discretisation time. The board gives no explicit + end-of-sweep marker, so this computed period is how "once per sweep" is timed + by default. + """ + if step <= 0: + raise ValueError(f"step must be > 0, got {step}") + span = abs(max_value - min_value) + num_steps = round(span / step) + 1 + per_point_s = delay_time_ms / 1000.0 + time_step_us / 1_000_000.0 + return num_steps * per_point_s + + +def resolve_period_s( + strategy: str, + *, + min_value: float, + max_value: float, + step: float, + time_step_us: float, + delay_time_ms: float, +) -> float: + """Turn a strategy string into a concrete per-reading period in seconds. + + Supported strategies: + - ``"computed"`` — one reading per estimated sweep duration (default). + - ``"interval:"`` — a fixed period of ```` milliseconds. + """ + if strategy == "computed": + return compute_sweep_period_s(min_value, max_value, step, time_step_us, delay_time_ms) + if strategy.startswith("interval:"): + try: + ms = float(strategy.split(":", 1)[1]) + except ValueError as exc: + raise ValueError(f"Invalid interval strategy {strategy!r}") from exc + if ms <= 0: + raise ValueError(f"interval must be > 0 ms, got {ms}") + return ms / 1000.0 + raise ValueError( + f"Unknown strategy {strategy!r}; expected 'computed' or 'interval:'" + ) + + +@dataclass(slots=True) +class LaserTemperatureMonitor: + """Polls a laser board once per sweep and publishes temperature readings. + + Args: + controller: object exposing ``get_measurements()`` (a real + :class:`LaserController` in production, a fake in tests). + writer: destination channel implementing ``write(TemperatureReading)``. + period_s: seconds between readings (see :func:`resolve_period_s`). + """ + + controller: _MeasurementSource + writer: ReadingWriter + period_s: float + + def read_once(self, seq: int) -> Optional[TemperatureReading]: + """Poll one measurement and turn it into a reading, or None if no data.""" + measurements = self.controller.get_measurements() + if measurements is None: + logger.warning("No measurement returned from laser board (seq=%d)", seq) + return None + return TemperatureReading( + seq=seq, + mono_ns=time.monotonic_ns(), + temp1=float(measurements.temp1), + temp2=float(measurements.temp2), + temp_ext1=_opt(getattr(measurements, "temp_ext1", None)), + temp_ext2=_opt(getattr(measurements, "temp_ext2", None)), + current1=_opt(getattr(measurements, "current1", None)), + current2=_opt(getattr(measurements, "current2", None)), + ) + + def run(self, stop_event: Optional[threading.Event] = None) -> None: + """Poll-and-publish until ``stop_event`` is set (runs forever if None). + + Each iteration reads once, publishes, then waits one period. The wait is + interruptible via ``stop_event`` for a prompt clean shutdown. + """ + stop = stop_event or threading.Event() + seq = 0 + logger.info("Temperature monitor started: period=%.3fs", self.period_s) + while not stop.is_set(): + try: + reading = self.read_once(seq) + except Exception: # noqa: BLE001 — a transient read error must not kill the monitor + logger.warning("Measurement read failed; continuing", exc_info=True) + reading = None + if reading is not None: + self.writer.write(reading) + logger.debug( + "Published reading seq=%d T1=%.3f T2=%.3f", seq, reading.temp1, reading.temp2 + ) + seq += 1 + stop.wait(self.period_s) + logger.info("Temperature monitor stopped after %d readings", seq) + + +def _opt(value: object) -> Optional[float]: + return None if value is None else float(value) diff --git a/python_app/hardware_full/laser_control/monitoring/readings_channel.py b/python_app/hardware_full/laser_control/monitoring/readings_channel.py new file mode 100644 index 0000000..26f21cc --- /dev/null +++ b/python_app/hardware_full/laser_control/monitoring/readings_channel.py @@ -0,0 +1,131 @@ +"""JSONL append/tail channel carrying per-sweep temperature readings. + +The monitor process appends one JSON object per line; the checker process tails +the file from its end and parses each newly-appended line. A newline-delimited +file is used (rather than a socket) so the monitor and checker can start, stop, +and restart on independent lifecycles without a handshake — the checker simply +resumes tailing wherever the file currently ends. + +A reader only ever consumes lines terminated by ``\\n``; a partially-written last +line is left buffered until its newline arrives, so a reading is never parsed +half-written. +""" + +from __future__ import annotations + +import json +import os +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Iterator, Optional, Union + +_PathLike = Union[str, os.PathLike[str]] + + +@dataclass(slots=True) +class TemperatureReading: + """One temperature/current snapshot published once per sweep. + + ``temp1``/``temp2`` are the internal laser temperatures (the values validated + against the setpoints); ``temp_ext1``/``temp_ext2`` are the external + thermistors, carried for diagnostics only. + """ + + seq: int + mono_ns: int + temp1: float + temp2: float + temp_ext1: Optional[float] = None + temp_ext2: Optional[float] = None + current1: Optional[float] = None + current2: Optional[float] = None + + def to_json_line(self) -> str: + return json.dumps(asdict(self), separators=(",", ":")) + + @classmethod + def from_json_line(cls, line: str) -> "TemperatureReading": + payload = json.loads(line) + return cls( + seq=int(payload["seq"]), + mono_ns=int(payload["mono_ns"]), + temp1=float(payload["temp1"]), + temp2=float(payload["temp2"]), + temp_ext1=_opt_float(payload.get("temp_ext1")), + temp_ext2=_opt_float(payload.get("temp_ext2")), + current1=_opt_float(payload.get("current1")), + current2=_opt_float(payload.get("current2")), + ) + + +def _opt_float(value: object) -> Optional[float]: + return None if value is None else float(value) + + +class ReadingWriter: + """Appends :class:`TemperatureReading` objects to a JSONL file. + + Each write is a single line flushed to the OS so a tailing reader sees it + promptly. Use as a context manager or call :meth:`close` explicitly. + """ + + def __init__(self, path: _PathLike) -> None: + self.path = Path(path) + self.path.parent.mkdir(parents=True, exist_ok=True) + # Line-buffered append; each reading is one line. + self._fh = self.path.open("a", encoding="utf-8", buffering=1) + + def write(self, reading: TemperatureReading) -> None: + self._fh.write(reading.to_json_line() + "\n") + self._fh.flush() + + def close(self) -> None: + if not self._fh.closed: + self._fh.close() + + def __enter__(self) -> "ReadingWriter": + return self + + def __exit__(self, *_exc: object) -> None: + self.close() + + +class ReadingReader: + """Tails a JSONL readings file, yielding complete lines as they appear. + + ``from_start=False`` (default) begins at the current end of file, so the + checker validates readings produced from the moment it starts. Partial + trailing lines are buffered until their newline arrives. + """ + + def __init__(self, path: _PathLike, *, from_start: bool = False) -> None: + self.path = Path(path) + self._buffer = "" + self._pos = 0 + if not from_start and self.path.exists(): + self._pos = self.path.stat().st_size + + def poll(self) -> Iterator[TemperatureReading]: + """Yield every complete reading appended since the last poll. + + Malformed lines are skipped silently (a truncated/legacy line must not + crash a long-running checker); callers that care can validate seq gaps. + """ + if not self.path.exists(): + return + with self.path.open("r", encoding="utf-8") as fh: + fh.seek(self._pos) + chunk = fh.read() + self._pos = fh.tell() + if not chunk: + return + self._buffer += chunk + *complete, self._buffer = self._buffer.split("\n") + for line in complete: + line = line.strip() + if not line: + continue + try: + yield TemperatureReading.from_json_line(line) + except (ValueError, KeyError, TypeError): + continue diff --git a/python_app/hardware_full/laser_control/monitoring/session.py b/python_app/hardware_full/laser_control/monitoring/session.py new file mode 100644 index 0000000..71bf951 --- /dev/null +++ b/python_app/hardware_full/laser_control/monitoring/session.py @@ -0,0 +1,75 @@ +"""Variation-session snapshot shared between the temperature monitor and checker. + +When a current-variation task is started, the target static laser temperatures +(``static_temp1``/``static_temp2``) and the acceptable tolerance are frozen into a +small JSON file. The temperature checker reads this file to know what "correct" +means for the run, so it validates against the setpoints that were in effect *at +variation start* — independent of any later edits to the run config. + +Only current variation of laser 1 (``CHANGE_CURRENT_LD1``) is supported by the +firmware today; the session still records both laser targets because both +temperatures are held static during that task and both are validated. +""" + +from __future__ import annotations + +import json +import os +import tempfile +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Union + +_PathLike = Union[str, os.PathLike[str]] + +# Default IPC locations. Both are overridable via CLI/API so several runs can use +# distinct files. Kept in the system temp dir so no project state is polluted. +DEFAULT_SESSION_PATH = Path(tempfile.gettempdir()) / "laser_variation_session.json" +DEFAULT_READINGS_PATH = Path(tempfile.gettempdir()) / "laser_temp_readings.jsonl" + +DEFAULT_TOLERANCE_C = 0.03 + + +@dataclass(slots=True) +class LaserVariationSession: + """Target setpoints and tolerance frozen at variation start.""" + + variation_type: str + target_temp1: float + target_temp2: float + tolerance_c: float = DEFAULT_TOLERANCE_C + started_at_iso: str = "" + + def save(self, path: _PathLike = DEFAULT_SESSION_PATH) -> Path: + """Atomically write the session snapshot to ``path`` and return it. + + Writes to a temp file in the same directory then renames, so a concurrent + checker never observes a half-written file. + """ + dest = Path(path) + dest.parent.mkdir(parents=True, exist_ok=True) + tmp = dest.with_name(f"{dest.name}.{os.getpid()}.tmp") + tmp.write_text(json.dumps(asdict(self), indent=2), encoding="utf-8") + os.replace(tmp, dest) + return dest + + @classmethod + def load(cls, path: _PathLike = DEFAULT_SESSION_PATH) -> "LaserVariationSession": + """Load a session snapshot from ``path``. + + Raises FileNotFoundError if the file is absent and ValueError if it is not + a valid session object. + """ + payload = json.loads(Path(path).read_text(encoding="utf-8")) + if not isinstance(payload, dict): + raise ValueError(f"Session file must be a JSON object: {path}") + try: + return cls( + variation_type=str(payload["variation_type"]), + target_temp1=float(payload["target_temp1"]), + target_temp2=float(payload["target_temp2"]), + tolerance_c=float(payload.get("tolerance_c", DEFAULT_TOLERANCE_C)), + started_at_iso=str(payload.get("started_at_iso", "")), + ) + except (KeyError, TypeError, ValueError) as exc: + raise ValueError(f"Malformed session file {path}: {exc}") from exc diff --git a/python_app/models/run_config_codec.py b/python_app/models/run_config_codec.py index d9b8cea..2a0ea77 100644 --- a/python_app/models/run_config_codec.py +++ b/python_app/models/run_config_codec.py @@ -331,6 +331,11 @@ def run_config_from_dict(payload: dict[str, Any]) -> RunConfigModel: model.radar.laser_control.variation.delay_time = _read_int( laser_variation_payload, "delay_time", model.radar.laser_control.variation.delay_time ) + model.radar.laser_control.variation.temp_tolerance_c = _read_float( + laser_variation_payload, + "temp_tolerance_c", + model.radar.laser_control.variation.temp_tolerance_c, + ) load_switch_payload(port1_payload, model.output_switch) load_switch_payload(port2_payload, model.input_switch) @@ -540,6 +545,7 @@ def run_config_to_dict(model: RunConfigModel) -> dict[str, Any]: "step": model.radar.laser_control.variation.step, "time_step": model.radar.laser_control.variation.time_step, "delay_time": model.radar.laser_control.variation.delay_time, + "temp_tolerance_c": model.radar.laser_control.variation.temp_tolerance_c, }, }, "sweep": sweep_payload, diff --git a/python_app/models/run_config_schema.py b/python_app/models/run_config_schema.py index 2250d73..13d3dc7 100644 --- a/python_app/models/run_config_schema.py +++ b/python_app/models/run_config_schema.py @@ -116,6 +116,9 @@ class LaserVariationModeModel: step: float = 0.1 time_step: int = 20 delay_time: int = 3 + # Max allowed |measured - target| laser temperature before the temperature + # checker warns, °C. Applied independently to both lasers (temp1/temp2). + temp_tolerance_c: float = 0.03 @dataclass(slots=True) diff --git a/python_app/scripts/laser_temp_checker.py b/python_app/scripts/laser_temp_checker.py new file mode 100644 index 0000000..5151c15 --- /dev/null +++ b/python_app/scripts/laser_temp_checker.py @@ -0,0 +1,78 @@ +"""Standalone laser temperature checker process. + +Reads the target setpoints frozen at variation start and tails the JSONL readings +channel produced by the monitor. For every reading it validates both lasers and +prints a console warning whenever a measured temperature drifts from its target by +more than the tolerance (default 0.03 °C). Never touches the serial port, so it is +fully independent of the monitor and can be started/stopped at any time. + +Example:: + + python -m python_app.scripts.laser_temp_checker + python -m python_app.scripts.laser_temp_checker --session s.json --readings r.jsonl +""" + +from __future__ import annotations + +import argparse +import logging +import signal +import threading +from pathlib import Path + +from python_app.hardware_full.laser_control.monitoring import ( + DEFAULT_READINGS_PATH, + DEFAULT_SESSION_PATH, + LaserTemperatureChecker, + LaserVariationSession, + ReadingReader, +) + +logger = logging.getLogger("laser_temp_checker") + +_POLL_INTERVAL_S = 0.2 + + +def main() -> int: + parser = argparse.ArgumentParser(description="Validate laser temperature against setpoints") + parser.add_argument("--session", type=Path, default=DEFAULT_SESSION_PATH, + help="Session snapshot with target setpoints + tolerance") + parser.add_argument("--readings", type=Path, default=DEFAULT_READINGS_PATH, + help="JSONL readings channel to tail") + parser.add_argument("--tolerance", type=float, default=None, + help="Override tolerance in °C (default: from session)") + parser.add_argument("--reminder-every", type=int, default=0, + help="Repeat a warning every N readings while off target (0=off)") + parser.add_argument("--from-start", action="store_true", + help="Validate the whole readings file, not just new lines") + args = parser.parse_args() + + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s") + + session = LaserVariationSession.load(args.session) + if args.tolerance is not None: + session.tolerance_c = args.tolerance + checker = LaserTemperatureChecker.from_session(session, reminder_every=args.reminder_every) + logger.info( + "Checking against T1=%.3f T2=%.3f °C, tolerance ±%.3f °C (%s)", + session.target_temp1, session.target_temp2, session.tolerance_c, session.variation_type, + ) + + reader = ReadingReader(args.readings, from_start=args.from_start) + stop_event = threading.Event() + + def request_stop(_signum: int, _frame: object) -> None: + stop_event.set() + + signal.signal(signal.SIGINT, request_stop) + signal.signal(signal.SIGTERM, request_stop) + + while not stop_event.is_set(): + for reading in reader.poll(): + checker.process(reading) + stop_event.wait(_POLL_INTERVAL_S) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/python_app/scripts/laser_temp_monitor.py b/python_app/scripts/laser_temp_monitor.py new file mode 100644 index 0000000..b8a21e0 --- /dev/null +++ b/python_app/scripts/laser_temp_monitor.py @@ -0,0 +1,142 @@ +"""Standalone laser temperature monitor process. + +Owns the laser serial port, (optionally) starts a current-variation task, then +polls the board once per sweep and appends each reading to a JSONL channel that +the temperature checker tails. Runs until SIGINT/SIGTERM. + +Examples:: + + # Start LD1 current variation from a run config, then monitor: + python -m python_app.scripts.laser_temp_monitor --config run_config.json --start + + # Monitor a variation that is already running: + python -m python_app.scripts.laser_temp_monitor --config run_config.json +""" + +from __future__ import annotations + +import argparse +import logging +import signal +import threading +from datetime import datetime +from pathlib import Path + +from python_app.hardware_full.laser_control.controller import ( + DEVICE_MAIN_MESSAGE_ID, + LaserController, +) +from python_app.hardware_full.laser_control.models import VariationType +from python_app.hardware_full.laser_control.monitoring import ( + DEFAULT_READINGS_PATH, + DEFAULT_SESSION_PATH, + LaserTemperatureMonitor, + LaserVariationSession, + ReadingWriter, + resolve_period_s, +) +from python_app.models.run_config_model import RunConfigModel + +logger = logging.getLogger("laser_temp_monitor") + + +def _start_variation(controller: LaserController, variation) -> None: + """Send the CHANGE_CURRENT_LD1 task and freeze the session snapshot.""" + controller.reset() + controller.set_manual_mode( + temp1=variation.static_temp1, + temp2=variation.static_temp2, + current1=variation.static_current1, + current2=variation.static_current2, + message_id=DEVICE_MAIN_MESSAGE_ID, + ) + controller.start_variation( + variation_type=VariationType[variation.variation_type], + params={ + "static_temp1": variation.static_temp1, + "static_temp2": variation.static_temp2, + "static_current1": variation.static_current1, + "static_current2": variation.static_current2, + "min_value": variation.min_value, + "max_value": variation.max_value, + "step": variation.step, + "time_step": variation.time_step, + "delay_time": variation.delay_time, + }, + ) + + +def main() -> int: + parser = argparse.ArgumentParser(description="Poll laser temperature once per sweep") + parser.add_argument("--config", required=True, type=Path, help="Path to run_config.json") + parser.add_argument("--readings", type=Path, default=DEFAULT_READINGS_PATH, + help="JSONL readings channel to append to") + parser.add_argument("--session", type=Path, default=DEFAULT_SESSION_PATH, + help="Session snapshot path (written with --start)") + parser.add_argument("--strategy", default="computed", + help="'computed' (per sweep) or 'interval:'") + parser.add_argument("--start", action="store_true", + help="Send CHANGE_CURRENT_LD1 before monitoring") + args = parser.parse_args() + + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s") + + config = RunConfigModel.load_from_path(args.config) + laser = config.radar.laser_control + variation = laser.variation + if variation.variation_type != "CHANGE_CURRENT_LD1": + logger.warning( + "Only CHANGE_CURRENT_LD1 is supported by firmware; got %s", + variation.variation_type, + ) + + period_s = resolve_period_s( + args.strategy, + min_value=variation.min_value, + max_value=variation.max_value, + step=variation.step, + time_step_us=variation.time_step, + delay_time_ms=variation.delay_time, + ) + + stop_event = threading.Event() + + def request_stop(_signum: int, _frame: object) -> None: + stop_event.set() + + signal.signal(signal.SIGINT, request_stop) + signal.signal(signal.SIGTERM, request_stop) + + controller = LaserController( + port=laser.port or None, + pi_coeff1_p=laser.pi_coeff1_p, + pi_coeff1_i=laser.pi_coeff1_i, + pi_coeff2_p=laser.pi_coeff2_p, + pi_coeff2_i=laser.pi_coeff2_i, + ) + controller.connect() + try: + if args.start: + _start_variation(controller, variation) + LaserVariationSession( + variation_type=variation.variation_type, + target_temp1=variation.static_temp1, + target_temp2=variation.static_temp2, + tolerance_c=variation.temp_tolerance_c, + started_at_iso=datetime.now().isoformat(timespec="seconds"), + ).save(args.session) + logger.info("Started CHANGE_CURRENT_LD1 and wrote session %s", args.session) + + with ReadingWriter(args.readings) as writer: + monitor = LaserTemperatureMonitor( + controller=controller, writer=writer, period_s=period_s + ) + logger.info("Monitoring to %s (period=%.3fs)", args.readings, period_s) + monitor.run(stop_event) + finally: + controller.disconnect() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/python_app/tests/test_laser_temp_monitoring.py b/python_app/tests/test_laser_temp_monitoring.py new file mode 100644 index 0000000..54a2c0e --- /dev/null +++ b/python_app/tests/test_laser_temp_monitoring.py @@ -0,0 +1,223 @@ +"""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()