rm3100/logger.py

#!/usr/bin/env python3
"""Log RM3100 magnetometer data over a CH347 USB-I2C adapter.

Run ./setup.sh first to install the udev rule and create the virtualenv, then:

    ./.venv/bin/python logger.py --duration 60
    ./.venv/bin/python logger.py --scan-only

The chip samples on its own internal schedule (manual sections 5.7.2, 5.8.2), so
`sample_index` is a grid coordinate, not a count of reads. A late read returns the
*newer* measurement rather than a delayed one, so an unnoticed miss would silently
compress the time axis. Misses are therefore detected and written as explicit
placeholder rows, keeping the index contiguous and the gap visible; the run
continues and reports the total at the end.

Only irreducible facts are written: the sample count, the host clock, and the raw
counts. Chip time, tesla and magnitude are all reconstructed on load from the
header -- see capture.py.
"""

import argparse
import csv
import os
import queue
import sys
import textwrap
import threading
import time
from datetime import datetime

import ch347
import rm3100

CSV_FIELDS = ["sample_index", "system_time_unix", "x_raw", "y_raw", "z_raw",
              "warning"]

# Mains fundamental. Anything sampled below twice this folds the line onto
# signal irreversibly, which is the one error post-processing cannot undo.
MAINS_HZ = 60.0

CONSOLE_REFRESH_S = 0.05
FLUSH_INTERVAL_S = 0.5

# How long to let the writer finish its backlog once sampling has stopped. Far
# longer than a healthy writer needs; exceeding it means rows are lost, which is
# reported rather than waited out indefinitely.
WRITER_DRAIN_TIMEOUT_S = 30.0

# The sampling loop releases the GIL inside each USB transfer, then has to take
# it back. Python's default 5 ms switch interval means the writer thread can
# hold it for that long, stalling the sampler and inflating the DRDY bracket --
# measured at ~10 ms against a 6.67 ms budget at 150 Hz, while the same loop
# without a writer thread peaked at 3.1 ms. Handing off 10x more often costs
# negligible throughput (the writer is not CPU-bound) and keeps the stall well
# inside the bracket budget.
GIL_SWITCH_INTERVAL_S = 0.0005

# Thread priority. The sampling loop runs on the main thread and is the only
# latency-sensitive part; the writer merely formats a few fields per sample.
#
# Lowering the writer needs no privilege and is always done: only the *relative*
# priority matters, and on Linux nice is per-thread, so it does not touch the
# sampler. Raising the sampler needs CAP_SYS_NICE and is opt-in via
# --high-priority, which fails loudly rather than degrading silently -- asking
# for it and not getting it is worth knowing about.
SAMPLER_NICE = -10
WRITER_NICE = 10


def restore_ownership(path):
    """Hand a file created under sudo back to the invoking user.

    --high-priority needs root, and anything root writes stays root-owned --
    which earlier in this project produced captures the normal user could not
    rewrite. Undo that here so privilege is needed for scheduling and nothing
    else leaks from it.
    """
    if os.geteuid() != 0:
        return
    uid, gid = os.environ.get("SUDO_UID"), os.environ.get("SUDO_GID")
    if uid is None:
        return
    try:
        os.chown(path, int(uid), int(gid) if gid else -1)
    except OSError as exc:
        print(f"WARNING: could not hand {path} back to uid {uid}: {exc}",
              file=sys.stderr)


def set_thread_nice(value):
    """Renice the calling thread. Returns the nice actually in effect, or None.

    Linux threads are tasks, so PRIO_PROCESS with a native thread id applies to
    just this thread. Elsewhere this may be a no-op or affect the process, which
    is why failure is tolerated rather than fatal -- priority is an optimisation,
    not a correctness requirement.
    """
    try:
        tid = threading.get_native_id()
        os.setpriority(os.PRIO_PROCESS, tid, value)
        return os.getpriority(os.PRIO_PROCESS, tid)
    except (AttributeError, OSError):
        return None

# Miss detection uses the DRDY bracket -- the span between the last poll showing
# DRDY clear and the poll showing it set -- rather than the read-to-read
# interval.
#
# The bracket is exact where the interval is only a heuristic. If DRDY reads
# clear at t_c then every earlier measurement has already been read, since a
# results read is what clears it. Measurements complete one period apart, so a
# bracket narrower than one period can contain at most one completion, and DRDY
# going high proves it contained at least one. Exactly one, whatever the host
# was doing beforehand.
#
# The read-to-read interval cannot make that claim: measured on this rig it
# reaches 37 ms against a 28.8 ms period -- a 29% overshoot from host stalls
# alone -- while the bracket stays under 10 ms. Thresholding the interval
# therefore flags healthy captures as lossy.
#
# The bracket is compared against the period measured by calibrate_period(), not
# the table value: the latter is 6-9% out on this unit, which is enough to
# miscount grid points inside a bracket and slip the sample index.

# Bounded so a stalled writer degrades predictably instead of exhausting memory
# on a long run. Far above the depth a healthy writer ever reaches, so filling it
# means the writer is wedged rather than merely behind -- and the run then ends,
# because a row that never reaches the file would leave a hole in sample_index
# and capture.py rejects a non-contiguous index outright. Stopping keeps
# everything already written valid.
QUEUE_MAX = 200_000

_SENTINEL = object()


# Rows carry flags in a `warning` column, space separated, empty when fine.
# Keeping them out of the numeric columns means x/y/z stay parseable as numbers
# (blank for a row with no data) and the column generalises to future flags.
#
# MISSED     this row is a placeholder -- the measurement was never read, so it
#            has no data. The row exists to keep sample_index contiguous, so it
#            remains a valid chip-time grid coordinate and the gap is explicit
#            rather than silently compressing the timeline.
# AMBIGUOUS  the gap ending at this row was of uncertain length, so the index
#            may have slipped from here on.
#
# The two are independent. A gap measuring 1.35 periods rounds to one, so no
# placeholder is written -- yet it sits far enough from an integer to distrust,
# and that case carries AMBIGUOUS on the *real* sample terminating the gap,
# which keeps its data. A placeholder inside an uncertain gap carries both.
#
# The flag is a confidence measure, not a claim that 1.35 and 1.65 are equally
# likely: 1.35 probably is one period and 1.65 probably two. Deciding which
# needs neighbouring timestamps and assumptions, and can still be wrong, which
# is exactly why the rows are flagged rather than silently resolved.
WARN_MISSED = "MISSED"
WARN_AMBIGUOUS = "AMBIGUOUS"

# How near a half-period the rounding may fall before the count is a coin toss.
AMBIGUITY_MARGIN = 0.25


def parse_args():
    p = argparse.ArgumentParser(description=__doc__,
                                formatter_class=argparse.RawDescriptionHelpFormatter)
    knob = p.add_mutually_exclusive_group()
    knob.add_argument("--rate", type=float, default=None,
                      help="target sample rate in Hz. The cycle count and TMRC "
                           "are derived from it, which is usually what you want "
                           "-- the cycle count is a continuous rate knob where "
                           "TMRC offers only factor-of-two steps")
    knob.add_argument("--cycle-count", type=int, default=None,
                      help="cycle count per axis, setting both the rate and the "
                           "resolution. Lower is faster but more coarsely "
                           f"quantised (default: {rm3100.DEFAULT_CYCLE_COUNT})")
    p.add_argument("--tmrc", type=lambda s: int(s, 0), default=None,
                   help="continuous-mode rate register, 0x92 (fastest) to 0x9D "
                        "(slowest). Default is the fastest, which lets the cycle "
                        "count set the rate -- give one only to sample slower "
                        "than the cycle count allows")
    p.add_argument("--duration", type=float, default=0.0,
                   help="seconds to log, or 0 to run until Ctrl-C (default: %(default)s)")
    p.add_argument("--output", default=None,
                   help="CSV output path (default: rm3100_<timestamp>.csv)")
    p.add_argument("--address", type=lambda s: int(s, 0), default=None,
                   help="I2C address, skipping the scan (default: autodetect)")
    p.add_argument("--bus-speed", type=int, choices=sorted(ch347.SPEEDS),
                   default=ch347.DEFAULT_SPEED_KHZ,
                   help="I2C bus speed in kHz. The default cycle count runs near "
                        "282 Hz, where 100 kHz would spend 42%% of each period on "
                        "the bus; 750 spends 6%% (default: %(default)s)")
    p.add_argument("--calibrate", type=float, default=1.0,
                   help="seconds of loss-free samples used to measure the true "
                        "measurement period before recording starts; the run "
                        "aborts if no clean stretch can be found (default: "
                        "%(default)s)")
    p.add_argument("--note", default=None,
                   help="free-text label recorded in the capture header, e.g. "
                        "the supply under test. Keeps the configuration with the "
                        "data instead of only in the filename")
    p.add_argument("--high-priority", action="store_true",
                   help=f"raise the sampling thread to nice {SAMPLER_NICE}. "
                        "Needs CAP_SYS_NICE, so run under sudo; the run aborts "
                        "if the priority cannot be set. Output files are handed "
                        "back to the invoking user afterwards")
    p.add_argument("--scan-only", action="store_true",
                   help="scan the bus, report what responded, and exit")
    return p.parse_args()


def print_plan(cfg, bus_speed, requested_rate, sampler_nice=None):
    """Show how the configuration was derived, so it can be checked not trusted."""
    lsb = rm3100.tesla_per_count(cfg.cycle_count) * rm3100.NT_PER_TESLA
    dither = rm3100.expected_noise_nt(cfg.cycle_count) / lsb
    period = 1.0 / cfg.predicted_hz
    bus = i2c_bus_time(bus_speed)
    share = bus / period

    print("Configuration")
    if requested_rate is not None:
        print(f"  requested       {requested_rate:g} Hz")
    print(f"  cycle count     {cfg.cycle_count:,}"
          f"{'':<8}1 / (3 x ({cfg.cycle_count}/"
          f"{rm3100.COUNTS_PER_SECOND:,.0f} + "
          f"{rm3100.AXIS_OVERHEAD_S * 1e6:.1f} us))")
    print(f"  sensor ceiling  {1 / rm3100.sample_period(cfg.cycle_count):.2f} Hz")
    print(f"  TMRC            0x{cfg.tmrc:02x} = "
          f"{rm3100.TMRC_RATES[cfg.tmrc]:g} Hz -- {cfg.governed_by} governs")
    print(f"  predicted rate  {cfg.predicted_hz:.2f} Hz "
          f"+/-{rm3100.RATE_TOLERANCE:.0%} (oscillator tolerance; measured below)")
    print(f"  resolution      {lsb:.2f} nT/LSB    dither {dither:.2f} LSB "
          "at spec noise")
    print(f"  duty            {cfg.duty:.1%}          integration / period")
    print(f"  bus {bus_speed} kHz    {bus * 1e3:.3f} ms/sample   "
          f"{share:.1%} of the period")
    if sampler_nice is not None:
        detail = (f"nice {sampler_nice}, writer at {WRITER_NICE}"
                  if sampler_nice < 0 else
                  f"nice {sampler_nice}, writer at {WRITER_NICE} "
                  "(--high-priority raises the sampler, needs sudo)")
        print(f"  priority        {detail}")
    for note in cfg.notes:
        print(f"  note: {note}")

    # Nothing below is ever fixed silently: changing a setting the user asked
    # for would hide the problem behind a configuration change. Say what is
    # wrong and what would fix it, then run what was requested.
    warnings = []

    # Below the recommended cycle count the LSB grows faster than the sensor's
    # own noise, so the quantiser stops being dithered and averaging stalls.
    # The manual's hard floor is 30 (section 5.1), but the margin is already
    # slim at 50, so anything under that is worth saying out loud.
    if cfg.cycle_count < rm3100.RECOMMENDED_MIN_CYCLE_COUNT:
        warnings.append(
            f"cycle count {cfg.cycle_count} is below the recommended "
            f"{rm3100.RECOMMENDED_MIN_CYCLE_COUNT}: dither is {dither:.2f} LSB "
            f"at spec noise ({lsb:.1f} nT/LSB) against the ~0.2 where averaging "
            "stops recovering sub-LSB resolution. characterize.py prints sd/LSB "
            "per axis, which settles it from the first capture.")

    # The measured worst configuration mistake -- TMRC slower than the
    # cycle-count ceiling leaves the sensor idle, and idle time buys nothing.
    # Idle is the shortfall against the ceiling, which is what TMRC costs; it is
    # not 1 - duty, since the fixed per-axis overhead is active time too.
    idle = 1.0 - cfg.predicted_hz * rm3100.sample_period(cfg.cycle_count)
    if cfg.governed_by == "TMRC" and idle > 0.2:
        warnings.append(
            f"TMRC 0x{cfg.tmrc:02x} governs and leaves the sensor idle "
            f"{idle:.0%} of each period, holding duty to {cfg.duty:.0%}. "
            "Measured cost at 23% duty was 1.43x the noise ASD. Raising the "
            "cycle count instead reaches the same rate at ~100% duty.")

    # A rate that came back different from the one asked for is easy to miss in
    # a note, and every figure above is derived from the rate.
    if (requested_rate is not None and abs(cfg.predicted_hz / requested_rate - 1)
            > rm3100.RATE_TOLERANCE):
        warnings.append(
            f"{requested_rate:g} Hz was requested but this configuration runs "
            f"at {cfg.predicted_hz:.2f} Hz.")

    # Mains and its harmonics fold onto signal below 2x mains, and no amount of
    # post-filtering undoes an alias.
    if cfg.predicted_hz < 2 * MAINS_HZ:
        alias = abs(MAINS_HZ - round(MAINS_HZ / cfg.predicted_hz) * cfg.predicted_hz)
        warnings.append(
            f"Nyquist is {cfg.predicted_hz / 2:.1f} Hz, below {MAINS_HZ:g} Hz "
            f"mains: interference folds to {alias:.2f} Hz and cannot be removed "
            "afterwards. Sampling faster and decimating gives the same noise "
            "floor with the mains line still visible.")

    # Table 3-1 stops at 400, so the noise figure printed above is extrapolation
    # past that point rather than a specification.
    if cfg.cycle_count > rm3100.MAX_SPEC_CYCLE_COUNT:
        warnings.append(
            f"cycle count {cfg.cycle_count:,} is past the {rm3100.MAX_SPEC_CYCLE_COUNT} "
            "where Table 3-1 ends, so the spec noise and gain above are "
            "extrapolated, not specified.")

    if share > 0.5:
        faster = [s for s in sorted(ch347.SPEEDS) if i2c_bus_time(s) / period < 0.25]
        fix = f"; {faster[0]} kHz would fit" if faster else ""
        warnings.append(f"the bus needs {share:.0%} of every period{fix}.")

    for text in warnings:
        print(textwrap.fill(text, width=79, initial_indent="  WARNING: ",
                            subsequent_indent="           "), file=sys.stderr)


def find_sensor(bus, address):
    """Locate the RM3100, or exit with wiring guidance."""
    if address is not None:
        print(f"Using I2C address 0x{address:02x} (scan skipped)")
        return address

    candidates = [a for a in rm3100.RM3100.ADDRESSES if bus.probe(a)]
    if not candidates:
        print("No RM3100 responded at 0x20-0x23.", file=sys.stderr)
        others = bus.scan()
        if others:
            print("Other devices on the bus: "
                  + ", ".join(f"0x{a:02x}" for a in others), file=sys.stderr)
        else:
            print("Nothing responded anywhere on the bus. Check, in order:\n"
                  "  - I2CEN tied high (otherwise the chip stays in SPI mode)\n"
                  "  - AVDD/VDD powered, not just DVDD\n"
                  "  - SDA/SCL not swapped\n"
                  "  - SDA and SCL pull-up resistors present", file=sys.stderr)
        sys.exit(1)

    if len(candidates) > 1:
        print("Multiple devices responded at "
              + ", ".join(f"0x{a:02x}" for a in candidates)
              + "; use --address to pick one.", file=sys.stderr)
        sys.exit(1)

    print(f"Found a device at 0x{candidates[0]:02x}")
    return candidates[0]


def start_sensor(sensor, cycle_count, tmrc):
    """Identify and configure the sensor, printing each step. Returns REVID."""
    revid = sensor.revid()
    if revid == rm3100.EXPECTED_REVID:
        print(f"REVID 0x{revid:02x} -- RM3100 confirmed")
    else:
        print(f"WARNING: REVID 0x{revid:02x}, expected "
              f"0x{rm3100.EXPECTED_REVID:02x}", file=sys.stderr)

    sensor.set_cycle_counts(cycle_count)
    readback = sensor.get_cycle_counts()
    if readback != (cycle_count,) * 3:
        print(f"ERROR: cycle count read back as {readback}, expected "
              f"{(cycle_count,) * 3}", file=sys.stderr)
        sys.exit(1)
    lsb_nt = rm3100.tesla_per_count(cycle_count) * rm3100.NT_PER_TESLA
    print(f"Cycle counts set to {readback} -- 1 LSB = {lsb_nt:.2f} nT")

    # Raises if DRC1 did not take; exactly-once sampling depends on it.
    sensor.configure()
    sensor.set_rate(tmrc)
    sensor.start_cmm()
    print("Continuous measurement mode started")
    return revid


def write_header(handle, meta):
    for key, value in meta.items():
        handle.write(f"# {key}: {value}\n")


def writer_thread(q, path, meta, stats):
    """Drain samples to CSV and drive the console, off the sampling thread."""
    # Stand aside for the sampler: this thread is not latency-sensitive.
    set_thread_nice(WRITER_NICE)
    lsb_nt = float(meta["tesla_per_count"]) * rm3100.NT_PER_TESLA
    with open(path, "w", newline="") as handle:
        restore_ownership(path)
        write_header(handle, meta)
        out = csv.writer(handle)
        out.writerow(CSV_FIELDS)
        last_print = 0.0
        last_flush = time.monotonic()

        while True:
            item = q.get()
            if item is _SENTINEL:
                break
            index, system_time, cx, cy, cz, warning = item

            # 6 decimal places is microsecond resolution on an epoch value,
            # far finer than the ~1 ms bracket uncertainty on each timestamp.
            # Placeholder rows carry zeros rather than blanks: the warning
            # column already says the row has no data, so keeping x/y/z strictly
            # integer makes the file trivial to parse.
            out.writerow([index, f"{system_time:.6f}",
                          0 if cx is None else cx,
                          0 if cy is None else cy,
                          0 if cz is None else cz, warning])
            stats["rows"] += 1

            now = time.monotonic()
            if now - last_flush >= FLUSH_INTERVAL_S:
                handle.flush()
                last_flush = now
            if now - last_print >= CONSOLE_REFRESH_S and cx is not None:
                # Placeholders carry no field values, so the display holds the
                # last real reading rather than blanking or showing nonsense.
                last_print = now
                x, y, z = cx * lsb_nt, cy * lsb_nt, cz * lsb_nt
                print(f"\r{stats['rows']:,} samples ({stats['missed']} missed)"
                      f"  |  X {x:+10.1f}  |  Y {y:+10.1f}  |  Z {z:+10.1f}"
                      f"  |  T {(x * x + y * y + z * z) ** 0.5:10.1f} nT",
                      end="", flush=True)
        handle.flush()


def i2c_bus_time(bus_speed_khz):
    """Seconds of I2C traffic the bus carries per sample.

    A combined register read of n payload bytes clocks n+3 bytes -- write
    address, register, read address, payload -- at 9 bit-times each (8 data plus
    ACK), and spends roughly a bit-time on each START and STOP. One sample costs
    a STATUS poll (1 payload byte) plus a results read (9).

    This is the irreducible traffic, not bus occupancy: the loop polls
    continuously while waiting for DRDY, so raw occupancy approaches 100% and
    says nothing useful. What matters is this against the sample period.
    """
    bits = 0
    for payload in (1, 9):
        bits += (payload + 3) * 9 + 3      # 2 STARTs and a STOP
    return bits / (bus_speed_khz * 1000.0)


def calibrate_period(sensor, dt_nominal, seconds, attempts=3):
    """Measure the true measurement period from a clean stretch of samples.

    Miss detection needs the real period, not the table value: the RC oscillator
    runs up to 7% off (manual section 5.2.1) and this unit is ~6% slow, which is
    enough to miscount grid points inside a bracket and slip the sample index.

    Least-squares slope of read time against index. Over one second at 35 Hz
    (~35 points, ~0.5 ms jitter) that pins the period to roughly 0.03%, against
    the 6-9% error of the nominal value -- and 0.03% is far finer than the
    resolution needed to tell k grid points from k+1.

    Requires a stretch with no miss, since an interval spanning a miss is a
    multiple of the period, not one period. Returns None if no clean stretch is
    available, which is itself the answer: the rate is not sustainable.

    Also returns the mean host cost of one sample's traffic -- a DRDY poll plus
    a results read -- timed here at no extra cost, since those are the very
    transactions being performed.
    """
    poll_ready, read_raw = sensor.poll_ready, sensor.read_raw
    monotonic = time.monotonic

    # Bootstrap the period from observation rather than trusting the TMRC table.
    # When the cycle count governs instead of TMRC -- which it does whenever the
    # cycle-count ceiling falls below the requested rate (manual section 5.2.1)
    # -- the table value is not merely 7% out but wrong by a large factor. At
    # cc=400 with TMRC 0x92 the table says 1.67 ms and the truth is 13.5 ms, so
    # a threshold built on it rejects every interval and calibration can never
    # start. A handful of raw intervals settles it: the median is robust to the
    # occasional stall, and only needs to be close enough to seed the real fit.
    probe = []
    clear = monotonic()
    deadline = clear + max(seconds, 0.25)
    while monotonic() < deadline and len(probe) < 24:
        ready_now, stamp, _ = poll_ready()
        if not ready_now:
            continue
        read_raw()
        probe.append(stamp)
    if len(probe) >= 4:
        gaps = sorted(b - a for a, b in zip(probe, probe[1:]))
        dt_nominal = gaps[len(gaps) // 2]

    for _ in range(attempts):
        times = []
        poll_cost = read_cost = 0.0
        polls = 0
        clear = monotonic()
        deadline = clear + seconds
        clean = True
        while monotonic() < deadline:
            before = monotonic()
            ready_now, after, _ = poll_ready()
            poll_cost += monotonic() - before
            polls += 1
            if not ready_now:
                clear = after
                continue
            ready = after
            if ready - clear >= dt_nominal:      # conservative: nominal is short
                clean = False
                break
            # Timed from here, not from `ready`: `ready` is the instant the
            # STATUS reply landed on the wire, so measuring against it would
            # charge the tail of the poll transaction to the read as well.
            before = monotonic()
            read_raw()
            read_cost += monotonic() - before
            times.append(ready)
            clear = ready
        n = len(times)
        if not clean or n < 8:
            continue
        # slope of t against i for i = 0..n-1, where Sxx = n(n^2-1)/12.
        mean_i = (n - 1) / 2.0
        mean_t = sum(times) / n
        sxy = sum((i - mean_i) * (t - mean_t) for i, t in enumerate(times))
        return (sxy / (n * (n * n - 1) / 12.0),
                poll_cost / max(polls, 1) + read_cost / n)
    return None, None


def sample_loop(sensor, q, duration, dt_nominal, stats):
    """Read every measurement, recording placeholders for any that are lost.

    The timestamp recorded per sample is the midpoint of the bracket between the
    last poll showing DRDY clear and the poll showing it set: the measurement
    completed somewhere in that window, and the midpoint is the best estimate
    available without a hardware DRDY line.

    Ends the run rather than dropping a row if the queue fills: sample_index has
    to stay contiguous to remain a chip-time grid coordinate, so a truncated
    capture is worth more than a longer one with a hole in it.
    """
    read_raw = sensor.read_raw
    poll_ready = sensor.poll_ready
    monotonic = time.monotonic
    put = q.put_nowait

    # DRDY may already be set from whatever ran before -- BIST in particular
    # zeroes the result registers and leaves it high. Discard one reading so
    # sample 0 is a genuinely fresh measurement.
    if poll_ready()[0]:
        read_raw()

    deadline = monotonic() + duration if duration > 0 else float("inf")
    index = 0
    # Bracket lower bound: the most recent moment DRDY was seen clear.
    clear_mono, clear_wall = monotonic(), time.time()

    # dt_nominal here is the *calibrated* period from calibrate_period(), not
    # the table value. Keep refining it from clean intervals so the estimate
    # follows the oscillator's thermal drift (~2500 ppm measured over 13 h).
    dt_est = dt_nominal
    previous_mono = previous_wall = None
    DT_SMOOTHING = 0.02

    while True:
        if monotonic() >= deadline:
            return
        # The stamps come back with the STATUS reply itself, taken the instant
        # it landed rather than several frames later up here.
        ready, ready_mono, ready_wall = poll_ready()
        if not ready:
            clear_mono, clear_wall = ready_mono, ready_wall
            continue
        # Monotonic drives detection so an NTP step cannot fake a miss; the
        # wall clock is recorded so a real NTP correction stays visible.
        bracket = ready_mono - clear_mono
        if bracket > stats["max_bracket"]:
            stats["max_bracket"] = bracket

        # Counting lost measurements in two stages, because the two questions
        # have different best answers.
        #
        # "Did we lose any?" is settled by the bracket, and rigorously: a
        # bracket shorter than one period cannot contain two completions, so
        # nothing was lost regardless of timing precision.
        #
        # "How many?" is better answered by the interval since the previous
        # accepted sample. Each accepted read sits on a grid point, so that
        # interval is very close to an exact multiple of the period, and
        # rounding recovers the multiple. The bracket alone cannot -- it only
        # spans back to the last poll that saw DRDY clear, which discards where
        # the grid actually is.
        #
        # Precision: the period is calibrated to ~0.07% and each completion is
        # located to about half a poll interval (~0.2 ms, under 1% of a period).
        # Over a gap of k periods the total error is well under a tenth of a
        # period for any small k, so the rounding is unambiguous -- which is
        # the case that matters, since a long gap means the run is unusable
        # anyway.
        lost = 0
        uncertain = False
        if bracket >= dt_est and previous_mono is not None:
            periods = (ready_mono - previous_mono) / dt_est
            lost = max(0, round(periods) - 1)
            # Uncertainty is a property of the gap, not of the placeholders:
            # a gap rounding to zero losses can still be a coin toss, and then
            # there is no placeholder to carry the flag.
            uncertain = abs(periods - round(periods)) > AMBIGUITY_MARGIN
            if uncertain:
                stats["ambiguous"] += 1
        elif previous_mono is not None:
            # A clean interval is exactly one period, so it calibrates dt_est.
            # Only clean ones qualify: an interval spanning a miss is a multiple.
            dt_est += (ready_mono - previous_mono - dt_est) * DT_SMOOTHING
        if lost:
            stats["missed"] += lost
            # Placeholders sit on the grid -- one period after the previous
            # accepted sample, and so on -- because that is where the lost
            # measurements actually completed. Spreading them across the
            # bracket instead would bunch them at the end of the gap and skew
            # the rate fit.
            flags = (f"{WARN_MISSED} {WARN_AMBIGUOUS}" if uncertain
                     else WARN_MISSED)
            for k in range(lost):
                try:
                    put((index, previous_wall + dt_est * (k + 1),
                         None, None, None, flags))
                except queue.Full:
                    stats["truncated"] = True
                    return
                index += 1

        sample_wall = (clear_wall + ready_wall) / 2.0
        (cx, cy, cz), _, _ = read_raw()
        try:
            # An uncertain gap that produced no placeholder still has to be
            # flagged, so it rides on the real sample that ends it.
            put((index, sample_wall, cx, cy, cz,
                 WARN_AMBIGUOUS if uncertain and not lost else ""))
        except queue.Full:
            stats["truncated"] = True
            return
        index += 1
        previous_mono, previous_wall = ready_mono, sample_wall
        # The read just cleared DRDY, so this is a known-clear instant.
        clear_mono, clear_wall = ready_mono, ready_wall


def main():
    args = parse_args()
    sys.setswitchinterval(GIL_SWITCH_INTERVAL_S)
    sampler_nice = None
    if args.high_priority:
        sampler_nice = set_thread_nice(SAMPLER_NICE)
        if sampler_nice is None:
            sys.exit(
                f"--high-priority needs CAP_SYS_NICE to set nice "
                f"{SAMPLER_NICE}, and this process does not have it.\n"
                "Re-run under sudo, using the venv interpreter by absolute "
                "path:\n"
                f"    sudo {sys.executable} {' '.join(sys.argv)}\n"
                "Or drop the flag -- the writer thread already steps aside, "
                f"which is the half that needs no privilege.")
    else:
        try:
            sampler_nice = os.getpriority(os.PRIO_PROCESS,
                                          threading.get_native_id())
        except (AttributeError, OSError):
            pass

    if args.cycle_count is not None and not 1 <= args.cycle_count <= 0xFFFF:
        sys.exit(f"--cycle-count {args.cycle_count} outside 1..65535")
    if args.rate is not None and args.rate <= 0:
        sys.exit(f"--rate {args.rate:g} must be positive")
    if args.tmrc is not None and args.tmrc not in rm3100.TMRC_RATES:
        sys.exit(f"--tmrc 0x{args.tmrc:02x} is not a valid rate register value")

    cfg = rm3100.plan(rate_hz=args.rate, cycle_count=args.cycle_count,
                      tmrc=args.tmrc)
    args.cycle_count, args.tmrc = cfg.cycle_count, cfg.tmrc

    try:
        bus = ch347.CH347I2C(ch347.SPEEDS[args.bus_speed])
    except IOError as exc:
        sys.exit(str(exc))

    with bus:
        print(f"CH347 adapter opened, I2C at {args.bus_speed} kHz")

        if args.scan_only:
            found = bus.scan()
            print("Devices found: " + ", ".join(f"0x{a:02x}" for a in found)
                  if found else "No devices responded on the bus.")
            return 0

        address = find_sensor(bus, args.address)
        sensor = rm3100.RM3100(bus, address)
        try:
            revid = start_sensor(sensor, args.cycle_count, args.tmrc)
        except IOError as exc:
            sys.exit(str(exc))

        print()
        print_plan(cfg, args.bus_speed, args.rate, sampler_nice)
        print()
        nominal_rate = cfg.predicted_hz

        # Establish the real period before recording. Everything downstream --
        # miss detection, placeholder counting, the sample-index grid -- depends
        # on it, and the table value is 6-9% out on this unit.
        dt, host_cost = calibrate_period(sensor, 1.0 / nominal_rate,
                                         args.calibrate)
        if dt is None:
            sensor.stop_cmm()
            sys.exit(
                f"Could not find {args.calibrate:g} s of loss-free samples at "
                f"{nominal_rate:.1f} Hz.\n"
                "Without a clean stretch the true period cannot be measured, so "
                "misses cannot be\ncounted reliably and the sample index would "
                "not track chip time.\n"
                "Lower the rate (higher --tmrc) or raise --bus-speed.")
        # Compared against the predicted rate, not TMRC's nominal: when the
        # cycle count governs, the TMRC figure is not what was aimed for and the
        # error against it is meaningless.
        print(f"Calibrated period {dt * 1e3:.4f} ms = {1 / dt:.4f} Hz "
              f"({(1 / dt) / cfg.predicted_hz - 1:+.2%} vs predicted)")
        # Whether the host is the constraint is a measurement, not a given: it
        # costs a near-constant ~0.9 ms per sample regardless of configuration,
        # so it dominates at short periods and disappears at long ones.
        share = host_cost / dt
        margin = ("little margin -- expect misses under load" if share > 0.5
                  else "modest margin" if share > 0.25
                  else "ample margin")
        print(f"Host cost {host_cost * 1e3:.3f} ms/sample "
              f"({share:.1%} of the period, {margin})")

        meta = {
            "rm3100_capture": 1,
            # The rate this configuration is predicted to produce, whichever of
            # the two ceilings governs -- not TMRC's table value, which is not
            # what was aimed for when the cycle count wins.
            "nominal_rate_hz": nominal_rate,
            "tmrc_nominal_hz": rm3100.TMRC_RATES[args.tmrc],
            "tmrc": f"0x{args.tmrc:02x}",
            "cycle_count": args.cycle_count,
            # repr() so the constant round-trips through float64 exactly.
            "tesla_per_count": repr(rm3100.tesla_per_count(args.cycle_count)),
            "i2c_address": f"0x{address:02x}",
            "bus_speed_khz": args.bus_speed,
            "revid": f"0x{revid:02x}",
            # What the logger actually used for miss detection. Not recoverable
            # from the data, since capture.py fits the whole run rather than the
            # first second.
            "calibrated_period_s": repr(dt),
        }
        if args.note:
            # Newlines would break the one-line-per-key header format.
            meta["note"] = " ".join(args.note.split())

        path = args.output or datetime.now().strftime("rm3100_%Y%m%d_%H%M%S.csv")
        print(f"Logging to {path} -- Ctrl-C to stop\n")

        q = queue.Queue(maxsize=QUEUE_MAX)
        stats = {"rows": 0, "missed": 0, "ambiguous": 0, "max_bracket": 0.0,
                 "truncated": False}
        writer = threading.Thread(target=writer_thread,
                                  args=(q, path, meta, stats), daemon=True)
        writer.start()

        try:
            sample_loop(sensor, q, args.duration, dt, stats)
        except KeyboardInterrupt:
            pass
        finally:
            q.put(_SENTINEL)
            writer.join(timeout=WRITER_DRAIN_TIMEOUT_S)
            drained = not writer.is_alive()
            print()
            try:
                sensor.stop_cmm()
                print("Continuous measurement mode stopped")
            except IOError as exc:
                print(f"WARNING: could not stop CMM: {exc}", file=sys.stderr)

        print(f"Wrote {stats['rows']} samples to {path}")
        # How close the run came to losing a measurement: the useful number for
        # judging whether a rate is sustainable before committing to a long run.
        # Quoted against the calibrated period, which is the threshold actually
        # enforced -- the nominal one is 6-9% out.
        print(f"Worst DRDY bracket {stats['max_bracket'] * 1e3:.2f} ms of "
              f"{dt * 1e3:.2f} ms allowed "
              f"({stats['max_bracket'] / dt * 100:.0f}% of margin used)")

        # Every one of these means the capture is usable but compromised in a
        # way the file alone does not show, so each is said out loud and each
        # makes the exit status non-zero.
        failed = False

        if stats["truncated"]:
            failed = True
            print("\nWARNING: recording stopped early -- the writer could not "
                  "keep up and the queue\nfilled. Continuing would have left a "
                  "hole in sample_index, which capture.py rejects,\nso the run "
                  "ended instead. Everything written is valid.", file=sys.stderr)

        if not drained:
            failed = True
            # The writer is a daemon thread, so it dies with the process and
            # anything still queued never reaches the file. Say so: a short tail
            # is not visible from the file itself.
            print(f"\nWARNING: the writer did not finish within "
                  f"{WRITER_DRAIN_TIMEOUT_S:g} s, so the last {q.qsize():,} "
                  "queued row(s) were\nnot written. The file is valid, just "
                  "short.", file=sys.stderr)

        if stats["ambiguous"]:
            failed = True
            print(f"\nWARNING: {stats['ambiguous']:,} gap(s) could not be "
                  "counted confidently -- the interval fell near a half-period, "
                  "so\nthe number of lost measurements is a guess and "
                  f"sample_index may have slipped.\nThose rows carry "
                  f"{WARN_AMBIGUOUS} in the warning column. Do not trust this "
                  "capture's\ntime axis for spectral work.", file=sys.stderr)

        if stats["missed"]:
            failed = True
            pct = stats["missed"] / max(stats["rows"], 1) * 100
            print(f"\nWARNING: {stats['missed']:,} measurement(s) were lost "
                  f"({pct:.3f}% of rows), flagged {WARN_MISSED}.\n"
                  "The sample index still tracks chip time -- the gaps are "
                  "explicit, not compressed --\nbut those rows carry no field "
                  "data. Lower the rate (higher --tmrc) or raise --bus-speed\n"
                  "to remove them.", file=sys.stderr)

        return 1 if failed else 0
    return 0


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