radar_system/python_app/scripts/laser_temp_monitor.py

"""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.exceptions import PortBusyError
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:<ms>'")
    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,
    )
    try:
        controller.connect()
    except PortBusyError as exc:
        # Expected, benign conflict: the manual-control UI (or another monitor)
        # already owns the port. Exit cleanly with guidance, not a traceback.
        logger.error("%s", exc)
        return 2
    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())