record_daq.py

#!/usr/bin/env python3
"""
record_daq.py -- record RAW spectra from a MAPIR DAQ light sensor (U / M / E)
to a Chloros-compatible .daq file, with NO MAPIR SDK and NO calibration.

This is a self-contained reference: it speaks the DAQ spectral-sensor wire
protocol directly over the three transports:

    DAQ-U : USB serial      (pyserial)
    DAQ-M : Bluetooth LE     (bleak, Nordic UART Service)
    DAQ-E : Ethernet         (stdlib sockets: JSON control + raw TCP)

By default it records the sensor's RAW firmware-output spectrum (no
calibration applied) and stamps the .daq so Chloros fetches this sensor's
factory calibration from the cloud BY SERIAL and applies it at import. You
capture; Chloros calibrates.

DAQ-E can also calibrate with no cloud and no Chloros at all. The device
carries its own factory bundle in flash, so ``--calibrate`` pulls it down
over ethernet and applies it locally (see daq_cal.py):

    --calibrate off    raw only. The default; unchanged behaviour.
    --calibrate csv    .daq stays RAW, plus a sibling .csv of calibrated
                       spectral irradiance in W/m^2/nm. Recommended: you get
                       usable numbers now and the .daq stays reprocessable.
    --calibrate bake   calibrated values go INTO the .daq, stamped with the
                       bundle sha. Self-contained, but frozen -- a future
                       bundle revision can no longer be applied to it.

Usage
-----
    python record_daq.py u --port COM7
    python record_daq.py u --port /dev/ttyUSB0 --integration-time 50 --frames 100
    python record_daq.py m --mac AA:BB:CC:DD:EE:FF
    python record_daq.py e --host 192.168.1.50 --duration 300
    python record_daq.py e --host 192.168.1.50 --calibrate csv

Press Ctrl-C to stop. Output defaults to ./<kind>_<timestamp>.daq.

Protocol summary (NSP32-style, all transports share it)
-------------------------------------------------------
  * Command:  03 BB <cmd> <user> [payload...] <checksum>
  * Response: 03 BB <cmd> <user> [payload...] <checksum>
  * checksum = ((~sum(bytes_before_checksum)) + 1) & 0xFF  (modular two's-comp;
    a valid packet has sum(whole_packet) & 0xFF == 0)
  * All multi-byte fields little-endian.
"""

import argparse
import csv
import io
import os
import signal
import socket
import struct
import sys
import threading
import time
from datetime import datetime, timezone

from mapir_metadata import DaqWriter, _VALID_KINDS

# "apply no per-wavelength profile at all" -- mirrors daq_cal.CAP_ID_AS_RECORDED
# and Chloros's daq.calibration_apply._CAP_ID_AS_RECORDED.
CAP_ID_AS_RECORDED = "as_recorded"

# DAQ-E JSON control channel (TCP 5001): reads the sensor's name (its
# calibration serial) and firmware at connect, and -- with --calibrate --
# pulls the onboard bundle and profiles.
#
# Shared with daq_cal rather than duplicated here: that implementation reads
# in 64 KB blocks rather than a byte at a time, which matters because
# get_calibration returns a single 5-30 KB line.
from daq_cal import DaqEControlClient as DaqEControl

# ---------------------------------------------------------------------------
# Wire protocol  (shared by all transports)
# ---------------------------------------------------------------------------
PFX0, PFX1 = 0x03, 0xBB
CMD_HELLO, CMD_STANDBY = 0x01, 0x04
CMD_GET_ID, CMD_GET_WL = 0x06, 0x24
CMD_ACQ, CMD_GET_SPEC = 0x26, 0x28

# Total length (bytes) of each RESPONSE packet, keyed by its command code.
# 135 spectral points: GetWavelength = 8 + 135*2 + 1 = 279;
# GetSpectrum = 12 + 135*4 + 12 + 1 = 565.
RET_LEN = {CMD_HELLO: 5, CMD_STANDBY: 5, CMD_GET_ID: 10,
           CMD_GET_WL: 279, CMD_ACQ: 5, CMD_GET_SPEC: 565}


def _checksum(data):
    """Modular two's-complement checksum placed as the last packet byte."""
    return ((~sum(data)) + 1) & 0xFF


def _checksum_ok(packet):
    return (sum(packet) & 0xFF) == 0


def _frame(code, user=0, payload=b""):
    body = bytearray([PFX0, PFX1, code & 0xFF, user & 0xFF])
    body += bytes(payload)
    body.append(_checksum(body))
    return bytes(body)


def cmd_hello():   return _frame(CMD_HELLO)
def cmd_standby(): return _frame(CMD_STANDBY)
def cmd_get_id():  return _frame(CMD_GET_ID)
def cmd_get_wl():  return _frame(CMD_GET_WL)


def cmd_acquire(integration_ms, frame_avg, enable_ae):
    """AcqSpectrum with 'active return' = 1: the sensor sends the GetSpectrum
    response automatically once the exposure completes."""
    payload = bytes([
        integration_ms & 0xFF, (integration_ms >> 8) & 0xFF,
        frame_avg & 0xFF, 1 if enable_ae else 0, 1])
    return _frame(CMD_ACQ, 0, payload)


def parse_sensor_id(packet):
    """GetSensorId response -> 'XX-XX-XX-XX-XX' (the calibration fetch key)."""
    return "-".join("%02X" % b for b in packet[4:9])


def parse_spectrum(packet):
    """GetSpectrum response -> (spectrum_floats, integration_ms, is_saturated).

    spectrum is the RAW firmware-output float array (pre-calibration) -- exactly
    what we store; Chloros applies the cloud bundle to it at import.
    """
    integration_ms = struct.unpack_from("<H", packet, 4)[0]
    is_saturated = packet[6] == 1
    n = struct.unpack_from("<I", packet, 8)[0]
    spectrum = struct.unpack_from("<%df" % n, packet, 12)
    return list(spectrum), integration_ms, is_saturated


# ---------------------------------------------------------------------------
# Stream framing  (serial + DAQ-E raw TCP: read one packet from a byte stream)
# ---------------------------------------------------------------------------
def read_stream_packet(read1, deadline):
    """Read+validate one protocol packet using read1() -> 1 byte (or b'').

    read1 should block up to its own short timeout and return b'' on timeout so
    we can honor the overall deadline. Returns the full packet bytes, or None on
    timeout / framing failure.
    """
    # sync on the 03 BB prefix
    prev = 0
    while True:
        if time.monotonic() > deadline:
            return None
        b = read1()
        if not b:
            continue
        if prev == PFX0 and b[0] == PFX1:
            break
        prev = b[0]
    # next byte is the command code -> tells us the total length
    cmd_b = _read_n(read1, 1, deadline)
    if not cmd_b:
        return None
    cmd = cmd_b[0]
    total = RET_LEN.get(cmd)
    if not total:
        return None  # unknown response; resync next call
    rest = _read_n(read1, total - 3, deadline)  # already consumed 03 BB cmd
    if rest is None or len(rest) != total - 3:
        return None
    packet = bytes([PFX0, PFX1, cmd]) + rest
    if not _checksum_ok(packet):
        return None
    return packet


def _read_n(read1, n, deadline):
    out = bytearray()
    while len(out) < n:
        if time.monotonic() > deadline:
            return None
        b = read1()
        if b:
            out += b
    return bytes(out)


# ---------------------------------------------------------------------------
# Transports
# ---------------------------------------------------------------------------
class SerialTransport:
    """DAQ-U: 115200 8N1 over a USB serial port (pyserial)."""

    def __init__(self, port):
        self.port = port
        self._ser = None

    def open(self):
        import serial  # pyserial
        self._ser = serial.Serial(
            self.port, baudrate=115200, bytesize=serial.EIGHTBITS,
            parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE,
            timeout=0.05)
        # drain any leftover bytes from a previous session
        try:
            self._ser.reset_input_buffer()
            self._ser.reset_output_buffer()
        except Exception:
            pass

    def send(self, data):
        self._ser.write(data)

    def read1(self):
        return self._ser.read(1)

    def recv_packet(self, timeout):
        return read_stream_packet(self.read1, time.monotonic() + timeout)

    def close(self):
        if self._ser is not None:
            try:
                self._ser.close()
            finally:
                self._ser = None


class TcpTransport:
    """DAQ-E raw channel: the same spectral protocol over a TCP socket."""

    def __init__(self, host, port=5000, timeout=2.0):
        self.host, self.port, self.timeout = host, port, timeout
        self._sock = None
        self._buf = bytearray()

    def open(self):
        self._sock = socket.create_connection((self.host, self.port),
                                              timeout=self.timeout)
        self._sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        self._sock.settimeout(0.05)

    def send(self, data):
        self._sock.sendall(data)

    def read1(self):
        if not self._buf:
            try:
                chunk = self._sock.recv(4096)
            except socket.timeout:
                return b""
            if not chunk:
                raise ConnectionError("DAQ-E closed the connection")
            self._buf.extend(chunk)
        b = bytes(self._buf[:1])
        del self._buf[:1]
        return b

    def recv_packet(self, timeout):
        return read_stream_packet(self.read1, time.monotonic() + timeout)

    def close(self):
        if self._sock is not None:
            try:
                self._sock.close()
            finally:
                self._sock = None


class BleTransport:
    """DAQ-M: Nordic UART Service over Bluetooth LE (bleak).

    Runs a private asyncio loop in a background thread so the recorder's main
    loop stays simple and synchronous. The RX-notify callback reassembles
    protocol packets and pushes complete ones onto a queue.
    """

    SERVICE = "6e400001-b5a3-f393-e0a9-e50e24dcca3e"
    TX_CHAR = "6e400002-b5a3-f393-e0a9-e50e24dcca3e"  # host -> device
    RX_CHAR = "6e400003-b5a3-f393-e0a9-e50e24dcca3e"  # device -> host (notify)

    def __init__(self, mac):
        self.mac = mac
        import queue
        self._pkts = queue.Queue()
        self._rx = bytearray()
        self._client = None
        self._loop = None
        self._thread = None
        self._ready = threading.Event()

    # -- background asyncio loop --
    def _run_loop(self):
        import asyncio
        self._loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self._loop)
        self._loop.run_forever()

    def _submit(self, coro, timeout=15.0):
        import asyncio
        fut = asyncio.run_coroutine_threadsafe(coro, self._loop)
        return fut.result(timeout=timeout)

    def open(self):
        from bleak import BleakClient
        self._thread = threading.Thread(target=self._run_loop, daemon=True)
        self._thread.start()
        while self._loop is None:
            time.sleep(0.01)
        self._client = BleakClient(self.mac)
        self._submit(self._client.connect(), timeout=30.0)
        self._submit(self._client.start_notify(self.RX_CHAR, self._on_notify))

    def _on_notify(self, _sender, data):
        # reassemble packets by the length-from-command-code rule
        self._rx += bytes(data)
        while len(self._rx) >= 3:
            cmd = self._rx[2]
            total = RET_LEN.get(cmd)
            if total is None:
                # unknown/garbage lead byte: drop one and resync
                del self._rx[0]
                continue
            if len(self._rx) < total:
                break
            pkt = bytes(self._rx[:total])
            del self._rx[:total]
            if _checksum_ok(pkt):
                self._pkts.put(pkt)

    def send(self, data):
        self._submit(self._client.write_gatt_char(self.TX_CHAR, data,
                                                  response=False))

    def recv_packet(self, timeout):
        import queue
        try:
            return self._pkts.get(timeout=timeout)
        except queue.Empty:
            return None

    def close(self):
        try:
            if self._client is not None:
                self._submit(self._client.disconnect(), timeout=10.0)
        except Exception:
            pass
        if self._loop is not None:
            self._loop.call_soon_threadsafe(self._loop.stop)


# (DaqEControl lives in daq_cal -- imported at the top of this module.)


# ---------------------------------------------------------------------------
# Sensor: connect handshake + acquire loop (transport-agnostic)
# ---------------------------------------------------------------------------
class DaqSensor:
    """Drives one sensor through a transport. The acquire loop sends
    AcqSpectrum (active-return) and waits for the GetSpectrum frame."""

    def __init__(self, kind, transport, *, integration_ms, frame_avg,
                 enable_ae):
        self.kind = kind                  # 'daq-u' / 'daq-m' / 'daq-e'
        self.t = transport
        self.integration_ms = integration_ms
        self.frame_avg = frame_avg
        self.enable_ae = enable_ae
        self.sensor_id = None
        self.fw = None
        # DAQ-E only: 'daq-e' or 'daq-e-s', straight from the hello. It
        # decides the cap default and it is NOT cosmetic -- a DAQ-E-S carries
        # its diffuser inside its own gain, so the sunshine profile a plain
        # DAQ-E needs would double-count it by ~11x.
        self.model = None

    def _exchange(self, command, want_cmd, timeout):
        """Send a command and return the next response with code want_cmd,
        skipping any intervening ACK packets (e.g. AcqSpectrum's 5-byte ACK)."""
        self.t.send(command)
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            pkt = self.t.recv_packet(max(0.05, deadline - time.monotonic()))
            if pkt is not None and pkt[2] == want_cmd:
                return pkt
        return None

    def connect(self):
        # DAQ-E: the serial number is the control-channel device name.
        if self.kind == "daq-e":
            self._control = self.t_control
            hello = self._control.cmd({"cmd": "hello"})
            self.fw = hello.get("fw")
            self.sensor_id = hello.get("name") or self.t.host
            # Read from hello rather than inferred from imu_present: a
            # DAQ-E-S whose IMU has failed would otherwise be misread as a
            # plain DAQ-E and silently given the wrong cap. Same source
            # Chloros uses (daq/sensors/e.py).
            self.model = hello.get("model") or None
        self.t.open()
        # Wake the sensor.
        self._exchange(cmd_hello(), CMD_HELLO, timeout=5.0)
        # DAQ-U / DAQ-M: serial number comes from GetSensorId.
        if self.kind in ("daq-u", "daq-m"):
            pkt = self._exchange(cmd_get_id(), CMD_GET_ID, timeout=5.0)
            if pkt is None:
                raise RuntimeError(f"{self.kind}: no GetSensorId response "
                                   f"(is the sensor connected/awake?)")
            self.sensor_id = parse_sensor_id(pkt)
        # Liveness: confirm the wavelength grid responds (not stored; Chloros
        # gets the grid from the calibration bundle at import).
        self._exchange(cmd_get_wl(), CMD_GET_WL, timeout=5.0)
        if not self.sensor_id:
            raise RuntimeError(f"{self.kind}: could not read a serial number")
        return self.sensor_id

    def read_spectrum(self):
        """Trigger one acquisition and return
        (spectrum_floats, integration_ms, is_saturated). Raw counts, no cal."""
        # exposure can take up to integration_time * frame_avg + overhead
        timeout = 2.0 + (self.integration_ms * max(self.frame_avg, 1)) / 1000.0 * 2
        pkt = self._exchange(
            cmd_acquire(self.integration_ms, self.frame_avg, self.enable_ae),
            CMD_GET_SPEC, timeout=timeout)
        if pkt is None:
            raise TimeoutError("no spectrum returned within timeout")
        return parse_spectrum(pkt)

    def standby_and_close(self):
        # Graceful shutdown: ask the sensor to stand by, then close the link.
        try:
            self.t.send(cmd_standby())
            time.sleep(0.1)
        except Exception:
            pass
        try:
            self.t.close()
        except Exception:
            pass
        if self.kind == "daq-e":
            try:
                self.t_control.close()
            except Exception:
                pass


# ---------------------------------------------------------------------------
# CLI / record loop
# ---------------------------------------------------------------------------
def build_sensor(args):
    kind = {"u": "daq-u", "m": "daq-m", "e": "daq-e"}[args.device]
    if kind == "daq-u":
        if not args.port:
            raise SystemExit("DAQ-U needs --port (e.g. COM7 or /dev/ttyUSB0)")
        transport = SerialTransport(args.port)
    elif kind == "daq-m":
        if not args.mac:
            raise SystemExit("DAQ-M needs --mac (the sensor's BLE address)")
        transport = BleTransport(args.mac)
    else:  # daq-e
        if not args.host:
            raise SystemExit("DAQ-E needs --host (the sensor's IP address)")
        transport = TcpTransport(args.host, port=args.raw_port)
    sensor = DaqSensor(kind, transport, integration_ms=args.integration_time,
                       frame_avg=args.frame_avg, enable_ae=not args.no_ae)
    if kind == "daq-e":
        sensor.t_control = DaqEControl(args.host, port=args.control_port,
                                       token=args.token or "")
    return sensor


def default_output(kind):
    # UTC, matching every other naive wall-clock these scripts write (the
    # capture TIFF filenames/EXIF and the .daq's stamped utc_offset_minutes=0
    # declaration). A local-time name here would misdescribe the recording's
    # declared timezone -- the exact mixed-clock bug the CM5 hub once had.
    stamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
    return f"{kind}_{stamp}.daq"


def main(argv=None):
    p = argparse.ArgumentParser(
        description="Record RAW spectra from a MAPIR DAQ-U/M/E to a "
                    "Chloros-compatible .daq (no SDK, no calibration).")
    p.add_argument("device", choices=["u", "m", "e"],
                   help="which DAQ: u=USB serial, m=Bluetooth LE, e=Ethernet")
    p.add_argument("--port", help="DAQ-U serial port (COM7, /dev/ttyUSB0)")
    p.add_argument("--mac", help="DAQ-M BLE address (AA:BB:CC:DD:EE:FF)")
    p.add_argument("--host", help="DAQ-E IP address")
    p.add_argument("--control-port", type=int, default=5001,
                   help="DAQ-E JSON control TCP port (default 5001)")
    p.add_argument("--token", default="",
                   help="DAQ-E control-channel auth token. A unit with one "
                        "provisioned (done at the calibration station) "
                        "refuses every command except hello without it -- "
                        "the symptom is auth_failed on connect or on "
                        "--calibrate's get_calibration. Harmless to omit on "
                        "unprovisioned units.")
    p.add_argument("--raw-port", type=int, default=5000,
                   help="DAQ-E raw spectral TCP port (default 5000)")
    p.add_argument("--integration-time", type=int, default=32,
                   help="integration time in ms (default 32)")
    p.add_argument("--frame-avg", type=int, default=3,
                   help="frames to average per reading (default 3)")
    p.add_argument("--no-ae", action="store_true",
                   help="disable auto-exposure (use fixed --integration-time)")
    p.add_argument("--frames", type=int, default=0,
                   help="stop after N readings (0 = until Ctrl-C)")
    p.add_argument("--duration", type=float, default=0.0,
                   help="stop after N seconds (0 = until Ctrl-C)")
    p.add_argument("--output", help="output .daq path (default <kind>_<ts>.daq)")
    p.add_argument("--device-name", default="", help="free-text label")
    p.add_argument("--calibrate", choices=["off", "csv", "bake"], default="off",
                   help="DAQ-E only: apply the bundle stored on the device "
                        "itself, with no cloud and no Chloros. 'csv' keeps the "
                        ".daq raw and writes calibrated W/m^2/nm alongside; "
                        "'bake' writes calibrated values into the .daq "
                        "(self-contained but no longer reprocessable). "
                        "Default 'off' (raw only).")
    p.add_argument("--csv", metavar="PATH", nargs="?", const=True,
                   help="also write a CSV of what this run records -- "
                        "calibrated W/m^2/nm with --calibrate csv, otherwise "
                        "the same RAW counts that go into the .daq. Bare "
                        "--csv puts it beside the .daq; give a PATH to place "
                        "it. Raw CSV is the only CSV a DAQ-U or DAQ-M can "
                        "produce, since neither carries a bundle to "
                        "calibrate against locally.")
    p.add_argument("--cap-id",
                   help="which cap is fitted. Default: the cap this model "
                        "ships wearing -- 'sunshine_cosine' for DAQ-U/M/E, "
                        "'as_recorded' for a DAQ-E-S. Pass 'none' ONLY for a "
                        "sensor you have physically stripped: declaring bare "
                        "on a capped unit reads 20-30x low at import and "
                        "nothing downstream can detect it. WITH --calibrate "
                        "this is also an override of the profile the device "
                        "reports, and the device carries ONE resolved "
                        "profile -- so there the only accepted values are "
                        "'as_recorded' (skip all per-wavelength profiles) or "
                        "the cap it already carries; naming a different one "
                        "is refused, because applying the stored curve under "
                        "another name is ~11x off between sunshine and bare. "
                        "To change caps, re-push profiles from Chloros.")
    p.add_argument("--require-profiles", action="store_true",
                   help="with --calibrate, fail unless the device carries cap/"
                        "geometry profiles. Use when a cap IS fitted: without "
                        "a profile the output is bare-uncorrected, which for a "
                        "sunshine cap is wrong by roughly 30x.")
    args = p.parse_args(argv)

    if args.calibrate != "off" and args.device != "e":
        raise SystemExit(
            "--calibrate needs the onboard bundle store, which only DAQ-E "
            "has. For DAQ-U / DAQ-M, record raw and let Chloros calibrate at "
            "import, or build a DeviceCalibration from a bundle file:\n"
            "    from daq_cal import DeviceCalibration\n"
            "    cal = DeviceCalibration.from_bundle(json.load(open(...)))")

    kind = {"u": "daq-u", "m": "daq-m", "e": "daq-e"}[args.device]
    sensor = build_sensor(args)

    stop = threading.Event()
    signal.signal(signal.SIGINT, lambda *_: stop.set())

    print(f"Connecting to {kind} ...", flush=True)
    serial_id = sensor.connect()
    print(f"  serial (cal key): {serial_id}"
          + (f"   fw: {sensor.fw}" if sensor.fw else "")
          + (f"   model: {sensor.model}" if sensor.model else ""), flush=True)

    # Offline calibration from the device's own flash. Reuses the already-open
    # control socket so we don't take a second of the DAQ-E's 4 control slots.
    cal = None
    if args.calibrate != "off":
        from daq_cal import DeviceCalibration
        cal = DeviceCalibration.from_device(
            args.host, control=sensor.t_control,
            require_profiles=args.require_profiles)
        print("Calibration (read from the device, no cloud):")
        for line in cal.describe().splitlines():
            print(f"  {line}")
        if args.cap_id:
            # Validate the override now rather than on the first frame: the
            # device carries one resolved profile, and naming a different cap
            # is refused (there is no local copy of its curve). Failing here
            # costs nothing; failing mid-recording loses the run.
            import numpy as _np
            try:
                cal.apply(_np.zeros(cal.n_points, dtype=_np.float32),
                          integration_time_ms=args.integration_time,
                          cap_id=args.cap_id)
            except Exception as e:
                raise SystemExit(f"--cap-id {args.cap_id}: {e}")
        if cal.profiles_source == "none" and not args.require_profiles:
            print("  ! no cap/geometry profiles on this device -- output is "
                  "bare-uncorrected.\n"
                  "    If a cap IS fitted this is wrong (a sunshine cap is "
                  "~30x). Connect the\n"
                  "    unit to Chloros once to push profiles, or re-run with "
                  "--require-profiles.",
                  file=sys.stderr, flush=True)

    out = args.output or default_output(kind)
    baking = (args.calibrate == "bake")

    # A DAQ-E-S is stamped as itself, not as a plain DAQ-E. Chloros maps both
    # to bundle kind 'e', but the cap treatment is opposite: a DAQ-E-S has its
    # diffuser inside its own gain and must get NO per-wavelength profile,
    # while a plain DAQ-E needs the sunshine curve (~11x). A file that cannot
    # tell them apart cannot be processed correctly.
    # Normalised and validated: `model` is whatever the firmware chose to put
    # in its hello, and an unrecognised string here would raise out of
    # DaqWriter AFTER the sensor is already streaming -- losing the run over a
    # spelling. Anything we do not recognise falls back to the --device kind
    # and says so, which is the pre-DAQ-E-S behaviour.
    product_model = kind
    if kind == "daq-e" and sensor.model:
        _m = str(sensor.model).strip().lower().replace("_", "-")
        if _m in _VALID_KINDS:
            product_model = _m
        else:
            print(f"  ! device reports model {sensor.model!r}, which this "
                  f"script does not recognise -- recording as {kind}. If this "
                  f"is a DAQ-E-S its cap default will be wrong; pass "
                  f"--cap-id as_recorded.", file=sys.stderr, flush=True)

    # Which cap Chloros should apply at import.
    #   explicit --cap-id            -> verbatim (the operator's statement)
    #   --calibrate reached the unit -> the profile the DEVICE carries, so the
    #                                   .daq and the sibling .csv agree
    #   otherwise                    -> None, and DaqWriter resolves the
    #                                   model's shipped state (sunshine for
    #                                   U/M/E, as_recorded for a DAQ-E-S).
    # NOT 'none': every unit MAPIR ships wears the sunshine corrector, and
    # declaring bare on a capped sensor is a 20-30x error at import that
    # nothing downstream can detect.
    cap_stamp = args.cap_id or (cal.cap_id if cal else None)

    # ...and WHO decided it. None lets DaqWriter resolve 'auto_default' (or
    # 'model' for a DAQ-E-S). This is what makes an assumed cap undoable
    # rather than merely wrong: Chloros warns on 'auto_default' and an
    # operator can override it per project, which it cannot do for a stamp
    # that claims someone checked.
    cap_source = ("operator" if args.cap_id
                  else "device" if cal else None)

    # cap_applied describes what is IN the blobs, so it has to follow the cap
    # actually handed to cal.apply() below -- not merely whether the device
    # had profiles. `--cap-id as_recorded` tells apply() to skip every
    # per-wavelength profile, so a recording made that way carries NO cap
    # however well stocked the device is; stamping 1 there claims a
    # correction that is not present, and Chloros then REFUSES a later
    # operator cap override (it cannot undo a curve that was never applied)
    # instead of correcting the file.
    cap_was_applied = (baking and cal is not None
                       and cal.profiles_source != "none"
                       and args.cap_id != CAP_ID_AS_RECORDED
                       and cap_stamp != CAP_ID_AS_RECORDED)

    writer = DaqWriter(
        out, product_model=product_model, product_serial=serial_id,
        device_name=args.device_name,
        cap_id=cap_stamp,
        cap_id_source=cap_source,
        calibration_applied=baking,
        calibration_bundle_sha=(cal.bundle_sha if baking else ""),
        calibration_completed_utc=(cal.completed_utc if baking else ""),
        cap_applied=cap_was_applied)

    csv_path = None
    csv_fh = None
    csv_out = None
    # A CSV of whatever this run is recording. With --calibrate csv that is
    # calibrated irradiance; otherwise it is the raw counts going into the
    # .daq -- which is the only CSV a DAQ-U or DAQ-M can produce at all,
    # since neither carries a bundle to calibrate against locally.
    #
    # The header says WHICH, in the file. Counts and W/m^2/nm differ by four
    # orders of magnitude and nothing about the numbers announces which you
    # are holding.
    want_csv = args.csv or args.calibrate == "csv"
    if want_csv:
        csv_calibrated = (args.calibrate == "csv")
        # Bare --csv (argparse const=True) means "beside the .daq".
        csv_path = (args.csv if isinstance(args.csv, str)
                    else os.path.splitext(out)[0] + ".csv")
        csv_fh = open(csv_path, "w", newline="", encoding="utf-8")
        csv_out = csv.writer(csv_fh)
        if csv_calibrated:
            csv_out.writerow([f"# MAPIR {kind} calibrated spectral irradiance "
                              f"(W/m^2/nm)"])
            csv_out.writerow([f"# sensor={serial_id} "
                              f"bundle_sha={cal.bundle_sha} "
                              f"completed={cal.completed_utc} cap={cal.cap_id} "
                              f"dark_model={cal.dark_model} "
                              f"profiles={cal.profiles_source}"])
        else:
            csv_out.writerow([f"# MAPIR {product_model} RAW sensor counts "
                              f"(NOT irradiance)"])
            # The cap is provenance for the .daq beside this file, not
            # something applied to these numbers -- say so, so nobody reads
            # the column as already-corrected.
            csv_out.writerow([f"# sensor={serial_id} uncalibrated; "
                              f"cap={cap_stamp} ({cap_source or 'assumed'}) "
                              f"is DECLARED for Chloros to apply, not applied "
                              f"here"])
        # Wavelength headers come from the bundle. A raw run has no
        # bundle, and the .daq deliberately does not store the axis
        # either (Chloros reconstructs it from the calibration), so a
        # raw CSV carries bare sample indices rather than inventing
        # nanometres it cannot know.
        _wl = ([f"{w:.1f}" for w in cal.wavelength_nm]
               if csv_calibrated else None)
        _csv_header_written = bool(_wl)
        if _wl:
            csv_out.writerow(["timestamp_utc", "timestamp_ns",
                              "integration_time_ms", "saturated",
                              "calibrated", "units"] + _wl)

    print(f"Recording {'CALIBRATED' if baking else 'RAW'} to: "
          f"{os.path.abspath(out)}")
    # State the cap, always. It is the one field an operator can get wrong
    # from the outside -- the sensor cannot sense what is screwed onto it --
    # and getting it wrong is 20-30x in downwelling, which is silent because
    # nothing downstream can check it against anything.
    _stamped = writer.cap_id
    _src = ("stated with --cap-id" if args.cap_id else
            "read from the device" if cal else
            f"ASSUMED -- the fleet default for {product_model}")
    print(f"  cap declared: {_stamped}  ({_src})")
    if cap_source is None and product_model != "daq-e-s":
        for _line in (
                "    The sunshine corrector is REMOVABLE on this model and the",
                "    sensor cannot tell whether it is fitted, so this is a",
                "    guess -- recorded as one (cap_id_source=auto_default), so",
                "    Chloros will flag it and you can override it later.",
                "    Pass --cap-id to state what is actually on the sensor."):
            print(_line)
    if _stamped == "none":
        for _line in (
                "  ! 'none' declares the cap PHYSICALLY REMOVED. MAPIR ships",
                "    these sensors with the sunshine corrector installed;",
                "    declaring bare on a capped unit reads 20-30x low at",
                "    import. Pass --cap-id sunshine_cosine if it is fitted."):
            print(_line, file=sys.stderr, flush=True)
    if csv_path:
        print(f"  calibrated CSV: {os.path.abspath(csv_path)}")
    if baking:
        print("  (.daq carries calibrated W/m^2/nm; Chloros will import "
              "as-is, not re-calibrate)")
    else:
        print("  (Chloros will fetch this sensor's calibration by serial "
              "at import)")
    print("  Ctrl-C to stop.", flush=True)

    t0 = time.monotonic()
    n = 0
    try:
        while not stop.is_set():
            try:
                spec, inttime, sat = sensor.read_spectrum()
            except TimeoutError as e:
                print(f"  ! {e}", file=sys.stderr, flush=True)
                continue
            ts = time.time_ns()
            calibrated = None
            if cal is not None:
                # Pass this frame's own integration time: auto-exposure moves
                # it between 1 and 500 ms and the dark model is a function of
                # it. Using the requested value instead of the reported one
                # would over-subtract at long integrations.
                calibrated = cal.apply(spec, integration_time_ms=inttime,
                                       cap_id=args.cap_id)
            writer.write(calibrated if baking else spec,
                         is_saturated=sat, integration_time_ms=inttime,
                         timestamp_ns=ts)
            if csv_out is not None:
                _vals = calibrated if csv_calibrated else spec
                if not _csv_header_written:
                    # Raw: the point count is only known once a
                    # frame has arrived, so the column row waits for
                    # one rather than guessing 135.
                    _csv_header_written = True
                    csv_out.writerow(
                        ["timestamp_utc", "timestamp_ns",
                         "integration_time_ms", "saturated",
                         "calibrated", "units"]
                        + [f"idx_{i}" for i in range(len(_vals))])
                csv_out.writerow(
                    [datetime.fromtimestamp(ts / 1e9,
                                            timezone.utc).isoformat(
                        timespec="milliseconds"),
                     ts, inttime, int(bool(sat)), int(csv_calibrated),
                     "W/m^2/nm" if csv_calibrated else "counts"]
                    + [f"{v:.6g}" for v in _vals])
            n += 1
            if n % 10 == 0:
                print(f"  {n} readings ...", flush=True)
            if args.frames and n >= args.frames:
                break
            if args.duration and (time.monotonic() - t0) >= args.duration:
                break
    finally:
        writer.close()
        if csv_fh is not None:
            csv_fh.close()
        sensor.standby_and_close()
        print(f"Stopped. Wrote {n} readings to {os.path.abspath(out)}")
        if csv_path:
            print(f"                    and {os.path.abspath(csv_path)}")
    return 0


if __name__ == "__main__":
    sys.exit(main())