#!/usr/bin/env python3 """ plot_fll_dac.py Companion PC-side tool for ReTr4's FllEngine (GnssDigitalDiscipline closed- loop OCXO discipline). Reads the sketch's EXISTING Serial "STAT ..." output as-is - no firmware changes required. FllEngine.cpp's fllPrintMachineStatus() already prints one STAT line per completed ~100s block (AVG_SECONDS), plus an occasional heartbeat backstop if PPS stalls, the entire time FLL is the active discipline - this script just listens, logs, and plots it. It does not touch the board. WHY THIS EXISTS: the FLL web status page's "DAC value, last N blocks" sparkline (MyWebServer.cpp) only keeps a fixed 40-entry RAM ring buffer (FllEngine.cpp's FLL_HISTORY_LEN) - about 67 minutes' worth - and is lost on reboot, and its bar heights are auto-scaled to their own local min/max, which can make small, legitimate single-DAC-count steps look like huge jumps when the whole visible window only spans a few counts. This script instead logs every STAT line, unbounded, with an explicit elapsed-time axis (t= from the sketch, seconds since boot) and exact numeric DAC/offset values, so an overnight (or longer) run can actually be inspected afterward rather than eyeballed from a screenshot of the last hour. This is meant purely as a temporary diagnostic while characterizing the fractional-remainder-accumulator change in GnssDigitalDiscipline (2026-09-20) - see the project handoff doc. The STAT line itself is NOT something this script adds or enables; FllEngine already prints it unconditionally whenever FLL is engaged, with no on/off switch today. If you later want the board itself to go quiet (e.g. for production use, once this line of debugging is done), that needs a small separate firmware change (a toggle around fllPrintMachineStatus()'s call sites) - ask if/when you want that; this script alone doesn't add or remove any board-side traffic, it only stops listening to it once you Ctrl+C. Parsed fields (all present on every well-formed STAT line from delivery 10 onward - see FllEngine.cpp's fllPrintMachineStatus() for the exact print sequence this mirrors; ldac/loff/lcnt are read with a default rather than required, so this script still works unmodified against an older delivery that doesn't print them yet): t (s since boot), mode, cyc (S/M/L/-), alarm (6-char R/P/F/D/_/H string), dac, avg (Hz, or nan before the first completed block), off (Hz, or nan), fix, sv, tacc, ok, wfix, wsv, wtacc, wok, pvts (this_stat/total), held (- or a reason string), blk (n/total - block-fill progress, NOT a running block index; resets every completion, so t= is the x-axis to use, not blk=), samples (total PIOFreqCounter samples seen since boot), ldac/loff/lcnt (GnssDigitalDiscipline's long-run average, added delivery 10 - "nan"/"nan"/0 until at least one LongCycle block has been accumulated; see GnssDigitalDiscipline.h's "LONG-RUN AVERAGE" comment). WHY ldac/loff MATTER FOR A MULTI-DAY RUN: the first 8h10m real-hardware run (2026-09-20) showed the instantaneous per-block DAC value can only ever move in +/-6/+/-7-count jumps at this OCXO's real gain/measurement-quantization combination (see the handoff doc's ninth entry) - never a clean single count. ldac is a SEPARATE, passive running mean of the DAC value actually in effect during every LongCycle ("locked") block, accumulated without lengthening AVG_SECONDS or slowing the loop's own reactivity at all. If the loop is genuinely dithering around a true optimum that sits between two adjacent DAC codes, ldac should converge, over many hours to days, toward a value with real sub-1-count precision (e.g. 2454.7, not just 2454 or 2455) - that convergence is the whole point of running this for days rather than hours, and is plotted as an overlay on the DAC panel below specifically so it's visible forming over the course of a long run rather than buried in a CSV column. Three logs are written per run: - : one row per STAT line, every field above plus a wall-clock timestamp and off_ppb (off_hz / nominal_hz * 1e9, matching the web page's own ppb display added 2026-09-20). - _y.txt: bare fractional-frequency values (off_hz/nominal_hz), one per line, no header - same minimal single-column convention as plot_frequency_counter.py's yfile, for a later Allan-deviation pass if wanted. - _events.txt: **NEW 2026-09-26** -- human-readable, timestamped log of serial-link disconnect/reconnect events only (see "RESILIENCE" below). Empty for a run with no drops. Kept SEPARATE from the CSV/yfile rather than interleaved as a special row, specifically so the yfile's strict one-bare-number-per-line convention (needed by a later Allan- deviation pass) is never at risk of being polluted by a non-numeric marker line. RESILIENCE (added 2026-09-26, after a real overnight run died on a dropped /dev/ttyACM0 -- see the project handoff doc): a multi-day unattended run must not die on a transient USB/serial hiccup. If ser.readline() raises (pyserial's "device reports readiness to read but returned no data (device disconnected or multiple access on port?)" is the one actually observed, but any SerialException/OSError from the read is treated the same way), this script now: logs a timestamped disconnect note to the events file, closes the stale port handle, and retries opening the port with capped exponential backoff (starts at RECONNECT_INITIAL_DELAY_S, doubles up to RECONNECT_MAX_DELAY_S) until it succeeds -- indefinitely; Ctrl+C still aborts at any point, including mid-retry. The CSV and yfile are NOT reopened/truncated across a reconnect -- same file handles, logging just resumes appending once the port is back. Bonus diagnostic: on the first STAT line parsed after a reconnect, this script compares its t= (board seconds-since-boot) against the last t= seen right before the drop. If the new t= is LOWER, the board's own uptime counter reset, i.e. the BOARD itself rebooted while the link was down. If the new t= is higher (continuing roughly where it left off, accounting for the downtime), the board's own uptime never reset -- only the HOST-side USB/serial link dropped, the board kept running the whole time. This is logged to the events file too, and is a direct, real- evidence way to distinguish "board reset" from "Linux/USB dropped the link" for whatever caused a given disconnect -- exactly the ambiguity a bare pyserial traceback leaves open. Live plot, two panels sharing the same elapsed-time x-axis, in HOURS since boot (t/3600 - changed 2026-09-21, OCXO timescales of interest are hours to days, not seconds; the CSV's own t_s column is still raw seconds, unaffected): 1. DAC value (raw counts) vs. elapsed hours, with the long-run mean DAC (ldac) overlaid as a second line once any LongCycle data exists - watch THIS line over a multi-day run, not the noisy instantaneous scatter. A RIGHT-hand secondary axis shows the same DAC-count scale converted to ppt (parts per trillion of OCXO_NOMINAL_HZ), RELATIVE TO THIS RUN'S OWN FIRST PLOTTED DAC VALUE (not an absolute dac=0 reference - see the 2026-09-21 fix note by the axis's own code, GnssDigitalDiscipline never operates anywhere near dac=0 in practice, so an absolute reference just buries the real, interesting variation under a large constant offset), so a step's size is directly readable as "how far has the equivalent pull moved since this plot started," in frequency terms - ticks formatted to 2 significant figures. Scale factor driven by --dac-full-scale-hz (see below). 2. Frequency offset (Hz) vs. elapsed hours Both panels mark any block whose cyc was S (ShortCycle) or M (MediumCycle) - i.e. NOT locked/LongCycle - with a distinct open marker (red triangle for S, orange square for M) layered on top of the regular scatter, right at the (t, value) point where it happened - added 2026-09-21 so acquisition bursts and real re-acquisition events (like the two large transients analyzed this session - see the handoff doc's eleventh/twelfth entries) are visually obvious without having to cross- reference the CSV's cyc column by hand. --dac-full-scale-hz: the board's REAL dacFullScaleHzGuess, needed only for the ppt secondary axis's scale factor. IMPORTANT: this is a RUNTIME value (FllEngine.cpp's fllSetup() loads it from flash if a CHARACTERIZE run has ever saved one - see the handoff doc's eleventh/twelfth entries) that can differ from - and, on this project's own bench unit, DOES differ from - AAconfig.h's OCXO_DAC_FULLSCALE_HZ compile-time default. Confirm the actual live value with the web status page or a bare 'R' over Serial (FllEngine.cpp prints "dacFullScaleHzGuess=...") rather than assuming; defaults here to -1.0100, this project's own bench unit's confirmed real value as of 2026-09-20 - override with --dac-full-scale-hz if you re-characterize, or if this is a different board. Usage: python3 plot_fll_dac.py --port /dev/ttyACM0 python3 plot_fll_dac.py --port COM5 --baud 115200 --csv run1.csv python3 plot_fll_dac.py --port /dev/ttyACM0 --dac-full-scale-hz -0.87 Dependencies (install once, on the machine you'll actually run this on - the one physically connected to the board over USB, not a remote/cloud shell): pip install pyserial matplotlib Ctrl+C to stop. The CSV and yfile are flushed after every line, so it's safe to interrupt at any point without losing already-logged data. """ import argparse import csv import sys import time from datetime import datetime # ---------------------------------------------------------------------- # Parsing: FllEngine.cpp's STAT line is a single, self-contained, space- # delimited "key=value" record per line (unlike baselineCounterEngine's # multi-line block report, so no multi-line accumulator state is needed # here - each line either parses to a complete record or it doesn't). # # IMPORTANT: split on whitespace FIRST, then split each token on the first # "=" - do NOT use a single blanket regex like r"(\w+)=(\S+)" over the # whole line. The held= field's value can itself contain embedded "="/"," # characters with no spaces, e.g. held=gnss(fix=1,tacc=50000,sv=3,ok=N) - # a whole-line regex would find those nested fix=/tacc=/sv=/ok= fragments # too and (since they appear later in the line) silently overwrite the # real top-level fix=/tacc=/sv=/ok= values with the ones from inside # held=. Token-then-split avoids this because every field the sketch # prints is one whitespace-delimited token with no internal spaces (see # fllPrintMachineStatus() in FllEngine.cpp) - held='s value has no spaces # either, so it survives as one token, it just isn't re-parsed for nested # "=" signs. # ---------------------------------------------------------------------- def parse_stat_line(line): """Returns a dict of field->string for a well-formed 'STAT ...' line, or None if this line isn't one (or is missing required fields).""" line = line.strip() if not line.startswith("STAT "): return None tokens = line.split()[1:] # drop the leading "STAT" token fields = {} for tok in tokens: if "=" not in tok: continue # malformed token - skip rather than raise, serial noise happens key, _, value = tok.partition("=") fields[key] = value required = {"t", "mode", "dac", "avg", "off", "blk", "samples"} if not required <= fields.keys(): return None return fields def to_float_or_none(s): if s is None or s.lower() == "nan": return None try: return float(s) except ValueError: return None # ---------------------------------------------------------------------- # Live serial + CSV + plot loop # ---------------------------------------------------------------------- DAC_FULLSCALE_COUNTS = 4095 # must match FllEngine.cpp's DAC_FULLSCALE (12-bit PWM DAC) # **NEW 2026-09-26** -- reconnect backoff for a dropped serial link (see the # "RESILIENCE" section of the module docstring). Starts short (a real USB # re-enumeration is usually done in well under a second) but backs off so a # genuinely unplugged/powered-off board doesn't spin a tight retry loop # forever; capped rather than unbounded so a long-absent board still gets # retried at a sane cadence rather than e.g. once a minute. RECONNECT_INITIAL_DELAY_S = 2.0 RECONNECT_MAX_DELAY_S = 30.0 CSV_COLUMNS = [ "timestamp", "t_s", "mode", "cyc", "alarm", "dac", "avg_hz", "off_hz", "off_ppb", "fix", "sv", "tacc", "ok", "wfix", "wsv", "wtacc", "wok", "pvts", "held", "blk", "samples", "long_run_mean_dac", "long_run_mean_off_hz", "long_run_count", ] def run(port, baud, csv_path, nominal_hz, yfile_path, dac_full_scale_hz, events_path): import serial # imported here so self_test() works without pyserial installed import matplotlib.pyplot as plt from matplotlib.ticker import FuncFormatter def _log_event(events_file, msg): """Timestamped line to both the console and the events file -- see the module docstring's RESILIENCE section. Kept as a tiny local helper since every call site wants both, flushed immediately (this may be the last thing written before the process dies for real).""" stamped = f"{datetime.now().isoformat(timespec='seconds')} {msg}" print(stamped) events_file.write(stamped + "\n") events_file.flush() def _open_serial(events_file, first_time): """Open the port, retrying with capped exponential backoff on failure. Blocks until successful (Ctrl+C aborts, same as any other wait in this script -- SerialException/OSError are the only things caught here, so KeyboardInterrupt still propagates normally). Used both for the initial open and for reconnecting after a drop.""" delay = RECONNECT_INITIAL_DELAY_S attempt = 0 while True: try: s = serial.Serial(port, baud, timeout=1) if not first_time: _log_event(events_file, f"reconnected to {port}") return s except (serial.SerialException, OSError) as e: attempt += 1 verb = "open" if first_time else "reopen" _log_event(events_file, f"failed to {verb} {port} (attempt {attempt}: {e}); " f"retrying in {delay:.0f}s") time.sleep(delay) delay = min(delay * 2, RECONNECT_MAX_DELAY_S) print(f"Logging to {csv_path}") print(f"Logging bare fractional-frequency values (one per line, no header) to {yfile_path}") print(f"Logging serial-link disconnect/reconnect events (if any) to {events_path}") # Opened before the port itself so that even a failed INITIAL open # (board not plugged in yet, wrong port name, etc.) gets logged rather # than just raising a bare traceback before anything else exists. events_file = open(events_path, "w") print(f"Opening {port} @ {baud} baud ...") ser = _open_serial(events_file, first_time=True) csv_file = open(csv_path, "w", newline="") writer = csv.writer(csv_file) writer.writerow(CSV_COLUMNS) csv_file.flush() yfile = open(yfile_path, "w") # **NEW 2026-09-26** -- last board-reported t= (seconds since boot) from # any successfully parsed STAT line, and whether the NEXT successfully # parsed line owes a continuity check against it (set True right after a # reconnect) -- see the module docstring's RESILIENCE section for what # this distinguishes (host-side USB drop vs. an actual board reboot). last_t_seen = None check_continuity_on_next = False # **CHANGED 2026-09-21** -- these hold ELAPSED HOURS (t_hr), not raw # seconds, for the plot's x-axis - OCXO timescales of interest are # hours/days; the CSV's own t_s column is unaffected (still seconds). t_hr_list, dac_vals, off_hz_vals = [], [], [] long_t_hr, long_dac_vals = [], [] # only populated once lcnt > 0 - see header comment # **NEW 2026-09-21** -- S/M ("not locked") block markers, tracked as # their own (t_hr, value) pairs rather than indexed off # t_hr_list/off_hz_vals, so there's no fragile alignment assumption # between series of different lengths (see the existing # off_hz_vals-may-trail-t_hr_list comment below). s_t_hr_dac, s_dac = [], [] m_t_hr_dac, m_dac = [], [] s_t_hr_off, s_off = [], [] m_t_hr_off, m_off = [], [] plt.ion() fig, (ax_dac, ax_off) = plt.subplots(2, 1, figsize=(9, 7), sharex=True) fig.suptitle("FLL DAC discipline (live)") line_dac, = ax_dac.plot([], [], marker="o", ms=3, lw=0.5, alpha=0.5, label="instantaneous DAC") line_ldac, = ax_dac.plot([], [], marker="", lw=2, color="crimson", label="long-run mean DAC (ldac)") marker_dac_s, = ax_dac.plot([], [], marker="^", ms=8, mfc="none", mec="tab:red", mew=1.5, ls="none", label="ShortCycle (S) block") marker_dac_m, = ax_dac.plot([], [], marker="s", ms=7, mfc="none", mec="darkorange", mew=1.5, ls="none", label="MediumCycle (M) block") ax_dac.set_ylabel("DAC value (counts)") ax_dac.legend(loc="best", fontsize=8) ax_dac.grid(True, alpha=0.3) # Secondary right-hand axis: same DAC-count scale, converted to ppt # (parts per trillion) of OCXO_NOMINAL_HZ via --dac-full-scale-hz. # # **FIXED 2026-09-21** -- the first version of this axis rescaled raw # DAC counts directly (counts * ppb_per_count), implicitly treating # dac=0 as the "0 ppb" reference point. That's fine mathematically, but # useless to read: this loop operates around dac~2400-2500 counts, not # 0, so EVERY tick showed the same ~-60 ppb "distance from dac=0" # baseline (2450 counts * ~-0.0247 ppb/count =~ -60.4 ppb) with the # actual, interesting step-to-step variation (~-0.025 ppb/count) buried # far past 2 significant figures underneath it - every label rounded to # the same "-60". Switching to ppt alone would NOT have fixed this (the # same swamping happens at any fixed linear scale, just with more # zeros: -60000 ppt every tick). The real fix is the reference point, # not the unit: this axis is now relative to THIS RUN's own first # plotted DAC value (ref_dac[0], set the first time a real sample # arrives, below) - "how far has the equivalent pull moved since this # plot started," which is what actually varies visibly, not "how far # from an arbitrary, physically-meaningless dac=0." hz_per_count = dac_full_scale_hz / DAC_FULLSCALE_COUNTS ppt_per_count = hz_per_count / nominal_hz * 1e12 ref_dac = [None] # set on the first real dac sample - see the loop below def _dac_to_ppt(counts): ref = ref_dac[0] if ref_dac[0] is not None else counts return (counts - ref) * ppt_per_count def _ppt_to_dac(ppt): ref = ref_dac[0] if ref_dac[0] is not None else 0.0 return ref + (ppt / ppt_per_count if ppt_per_count else 0.0) # **CHANGED 2026-09-21** -- shortened after the user reported the first # (long, "..., rel. to run start (@ dacFullScaleHz=...)") label getting # clipped/overlapping at the right edge of the window. The full detail # (what "rel. to" means, where --dac-full-scale-hz comes from) is in the # module docstring, not needed on the live axis label itself. ax_dac_ppt = ax_dac.secondary_yaxis("right", functions=(_dac_to_ppt, _ppt_to_dac)) ax_dac_ppt.set_ylabel("Equiv. pull, ppt (vs. start)") ax_dac_ppt.yaxis.set_major_formatter(FuncFormatter(lambda v, _pos: f"{v:.2g}")) line_off, = ax_off.plot([], [], marker="o", ms=3, color="tab:orange", label="offset") marker_off_s, = ax_off.plot([], [], marker="^", ms=8, mfc="none", mec="tab:red", mew=1.5, ls="none", label="ShortCycle (S) block") marker_off_m, = ax_off.plot([], [], marker="s", ms=7, mfc="none", mec="darkorange", mew=1.5, ls="none", label="MediumCycle (M) block") ax_off.axhline(0, color="gray", linewidth=0.8, linestyle="--") ax_off.set_ylabel("Frequency offset (Hz)") ax_off.set_xlabel("Elapsed time (hours since boot)") ax_off.legend(loc="best", fontsize=8) ax_off.grid(True, alpha=0.3) # Reserve room on the right for ax_dac_ppt's label/ticks - without this # the secondary axis's label can get clipped against the window edge # (the "cut-off text to the right" the user reported). fig.tight_layout() fig.subplots_adjust(right=0.87) print("Waiting for STAT lines (Ctrl+C to stop) ...") try: while True: try: raw = ser.readline() except (serial.SerialException, OSError) as e: # **NEW 2026-09-26** -- the drop this whole mechanism exists # for (see RESILIENCE in the module docstring). Don't lose # the run: log it, close the stale handle, reopen (blocks, # with backoff, until it succeeds or the user Ctrl+C's), and # pick up right where we left off -- same csv_file/yfile, # nothing truncated. _log_event(events_file, f"serial link dropped ({e}); reconnecting ...") try: ser.close() except Exception: pass # already gone -- nothing to clean up ser = _open_serial(events_file, first_time=False) check_continuity_on_next = True continue if not raw: continue try: line = raw.decode("utf-8", errors="replace").rstrip("\r\n") except Exception: continue if line: print(line) # echo raw sketch output too, useful while watching live fields = parse_stat_line(line) if fields is None: continue t = to_float_or_none(fields["t"]) # **NEW 2026-09-26** -- first STAT line parsed after a reconnect: # compare against the board's own uptime counter (t=) from just # before the drop. Lower now = the BOARD itself rebooted while # the link was down (its own t= restarted near 0). Higher/ # continuing now = only the HOST-side USB/serial link dropped; # the board's firmware never stopped running. Real, direct # evidence either way -- logged so it's not lost once the plot # window scrolls past it. if check_continuity_on_next and t is not None: if last_t_seen is not None: if t < last_t_seen: _log_event( events_file, f"board uptime (t=) RESET after reconnect: was {last_t_seen:.0f}s, " f"now {t:.0f}s -- the BOARD itself rebooted while the link was down, " f"this was not just a host-side USB drop") else: _log_event( events_file, f"board uptime (t=) CONTINUED after reconnect: was {last_t_seen:.0f}s, " f"now {t:.0f}s -- the board's firmware kept running the whole time, " f"this was a host-side USB/serial drop, not a board reboot") check_continuity_on_next = False if t is not None: last_t_seen = t # **NEW 2026-09-21** -- t_hr is ONLY for the plot's x-axis (OCXO # timescales of interest are hours/days, not seconds); the CSV's # own t_s column and everything else still uses raw t (seconds), # unchanged, matching the sketch's own t= field exactly. t_hr = (t / 3600.0) if t is not None else None dac = to_float_or_none(fields["dac"]) avg_hz = to_float_or_none(fields["avg"]) off_hz = to_float_or_none(fields["off"]) off_ppb = (off_hz / nominal_hz * 1e9) if off_hz is not None else None # ldac/loff/lcnt: absent entirely on an older (pre-delivery-10) # firmware, or "nan"/0 before any LongCycle block has been # accumulated yet - to_float_or_none()/int() handle both via # fields.get()'s default. long_run_mean_dac = to_float_or_none(fields.get("ldac")) long_run_mean_off_hz = to_float_or_none(fields.get("loff")) try: long_run_count = int(fields.get("lcnt", "0")) except ValueError: long_run_count = 0 cyc = fields.get("cyc", "") writer.writerow([ datetime.now().isoformat(timespec="seconds"), t, fields.get("mode", ""), fields.get("cyc", ""), fields.get("alarm", ""), dac, avg_hz, off_hz, off_ppb, fields.get("fix", ""), fields.get("sv", ""), fields.get("tacc", ""), fields.get("ok", ""), fields.get("wfix", ""), fields.get("wsv", ""), fields.get("wtacc", ""), fields.get("wok", ""), fields.get("pvts", ""), fields.get("held", ""), fields.get("blk", ""), fields.get("samples", ""), long_run_mean_dac, long_run_mean_off_hz, long_run_count, ]) csv_file.flush() if off_hz is not None: yfile.write(f"{off_hz / nominal_hz:.15e}\n") yfile.flush() if t_hr is not None and dac is not None: if ref_dac[0] is None: ref_dac[0] = dac # first real sample -> this run's ppt-axis zero point t_hr_list.append(t_hr) dac_vals.append(dac) line_dac.set_data(t_hr_list, dac_vals) ax_dac.relim(); ax_dac.autoscale_view() # **NEW 2026-09-21** -- mark this exact point if it was a # not-locked (S/M) block. Own (t_hr, value) pairs, not # indexed off t_hr_list, so this never assumes alignment # with any other series - see the tracking-list comment above. if cyc == "S": s_t_hr_dac.append(t_hr); s_dac.append(dac) marker_dac_s.set_data(s_t_hr_dac, s_dac) elif cyc == "M": m_t_hr_dac.append(t_hr); m_dac.append(dac) marker_dac_m.set_data(m_t_hr_dac, m_dac) if t_hr is not None and long_run_mean_dac is not None and long_run_count > 0: long_t_hr.append(t_hr) long_dac_vals.append(long_run_mean_dac) line_ldac.set_data(long_t_hr, long_dac_vals) ax_dac.relim(); ax_dac.autoscale_view() if t_hr is not None and off_hz is not None: off_hz_vals.append(off_hz) # off_hz_vals may trail t_hr_list by one entry if an early # STAT line had dac but avg=nan/off=nan (before the first # completed block) - only plot the pairs we actually have. line_off.set_data(t_hr_list[-len(off_hz_vals):], off_hz_vals) ax_off.relim(); ax_off.autoscale_view() if cyc == "S": s_t_hr_off.append(t_hr); s_off.append(off_hz) marker_off_s.set_data(s_t_hr_off, s_off) elif cyc == "M": m_t_hr_off.append(t_hr); m_off.append(off_hz) marker_off_m.set_data(m_t_hr_off, m_off) fig.canvas.draw() plt.pause(0.01) except KeyboardInterrupt: print("\nStopped.") finally: csv_file.close() yfile.close() events_file.close() try: ser.close() except Exception: pass # may already be closed/gone if we died mid-reconnect # ---------------------------------------------------------------------- # Self-test: verifies the parser against synthetic STAT lines matching # FllEngine.cpp's fllPrintMachineStatus() exact print sequence, including # the held=gnss(...) embedded-"=" case that motivated the token-then-split # parsing approach above. Run with --self-test - no serial port or # hardware needed. # ---------------------------------------------------------------------- SAMPLE_LINE_LOCKED = ( "STAT t=3812 mode=CLOSEDLOOP cyc=L alarm=r_f_d_h dac=2457 avg=9999999.9998 " "off=-0.0002 fix=3 sv=9 tacc=25000 ok=Y wfix=3 wsv=9 wtacc=27000 wok=Y " "pvts=98/3812 held=- blk=100/100 samples=3812 ldac=2454.7123 loff=0.000012 lcnt=214" ) SAMPLE_LINE_HELD_GNSS = ( "STAT t=920 mode=CLOSEDLOOP cyc=M alarm=r_f_d_h dac=2048 avg=10000000.4100 " "off=0.4100 fix=1 sv=3 tacc=90000 ok=N wfix=1 wsv=3 wtacc=90000 wok=N " "pvts=45/920 held=gnss(fix=1,tacc=90000,sv=3,ok=N) blk=100/100 samples=920 " "ldac=nan loff=nan lcnt=0" ) SAMPLE_LINE_PRE_FIRST_BLOCK = ( "STAT t=12 mode=CLOSEDLOOP cyc=- alarm=r_f_d_h dac=2048 avg=nan " "off=nan fix=0 sv=0 tacc=999999999 ok=N wfix=0 wsv=0 wtacc=999999999 wok=N " "pvts=12/12 held=- blk=12/100 samples=12 ldac=nan loff=nan lcnt=0" ) # A pre-delivery-10 STAT line, with NO ldac/loff/lcnt fields at all - the # parser and run() loop must still handle this gracefully (older firmware, # or a log captured before this feature existed). SAMPLE_LINE_NO_LONG_RUN_FIELDS = ( "STAT t=3812 mode=CLOSEDLOOP cyc=L alarm=r_f_d_h dac=2457 avg=9999999.9998 " "off=-0.0002 fix=3 sv=9 tacc=25000 ok=Y wfix=3 wsv=9 wtacc=27000 wok=Y " "pvts=98/3812 held=- blk=100/100 samples=3812" ) NOT_A_STAT_LINE = "FllEngine: setup" def self_test(): r = parse_stat_line(SAMPLE_LINE_LOCKED) assert r is not None assert r["t"] == "3812" assert r["dac"] == "2457" assert r["off"] == "-0.0002" assert r["held"] == "-" assert to_float_or_none(r["avg"]) == 9999999.9998 # The case that motivated token-then-split over a whole-line regex: # held='s embedded fix=/tacc=/sv=/ok= must NOT clobber the real # top-level fix=/tacc=/sv=/ok= values, which appear earlier on the line. r2 = parse_stat_line(SAMPLE_LINE_HELD_GNSS) assert r2 is not None assert r2["fix"] == "1" and r2["tacc"] == "90000" and r2["sv"] == "3" and r2["ok"] == "N", ( "top-level fix/tacc/sv/ok were clobbered by held='s embedded copies: " + repr(r2) ) assert r2["held"] == "gnss(fix=1,tacc=90000,sv=3,ok=N)" # **NEW 2026-09-21** -- cyc must parse cleanly for the S/M plot-marker # feature; SAMPLE_LINE_LOCKED (L) and SAMPLE_LINE_HELD_GNSS (M) cover # both the "don't mark" and "do mark" cases. assert r["cyc"] == "L" assert r2["cyc"] == "M" r3 = parse_stat_line(SAMPLE_LINE_PRE_FIRST_BLOCK) assert r3 is not None assert to_float_or_none(r3["avg"]) is None, "avg=nan should parse to None, not a float" assert to_float_or_none(r3["off"]) is None # ldac/loff/lcnt: present-and-numeric, present-and-nan, and absent # entirely (older firmware) must all parse without raising. assert to_float_or_none(r["ldac"]) == 2454.7123 assert to_float_or_none(r["loff"]) == 0.000012 assert int(r["lcnt"]) == 214 assert to_float_or_none(r2.get("ldac")) is None, "ldac=nan should parse to None, not a float" assert int(r2.get("lcnt", "0")) == 0 r4 = parse_stat_line(SAMPLE_LINE_NO_LONG_RUN_FIELDS) assert r4 is not None, "a pre-delivery-10 STAT line with no ldac/loff/lcnt must still parse" assert to_float_or_none(r4.get("ldac")) is None assert int(r4.get("lcnt", "0")) == 0 assert parse_stat_line(NOT_A_STAT_LINE) is None assert parse_stat_line("") is None print("self_test: OK -- parser correctly extracted all sample STAT lines, including the " "held=gnss(...) embedded-'=' case without clobbering top-level fields, and the " "ldac/loff/lcnt long-run-average fields (numeric, nan, and entirely absent):") for r in (r, r2, r3, r4): print(" ", r) def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--port", help="Serial port, e.g. /dev/ttyACM0 or COM5 (required unless --self-test)") ap.add_argument("--baud", type=int, default=115200) ap.add_argument("--csv", default=None, help="CSV log path (default: fll_dac_.csv)") ap.add_argument("--yfile", default=None, help="Bare single-column fractional-frequency log path, no header " "(default: _y.txt)") ap.add_argument("--nominal-hz", type=float, default=10_000_000.0, help="Must match OCXO_NOMINAL_HZ in AAconfig.h (default 10e6)") ap.add_argument("--dac-full-scale-hz", type=float, default=-1.0100, help="Board's REAL dacFullScaleHzGuess (check via the web status page or a " "bare 'R' over Serial - do NOT assume AAconfig.h's compile-time default, " "it can be overridden by a flash-persisted CHARACTERIZE result - see the " "module docstring). Only used to scale the DAC panel's right-hand ppt " "axis. Default -1.0100 is this project's own bench unit's confirmed real " "value as of 2026-09-20.") ap.add_argument("--self-test", action="store_true", help="Run the parser self-test and exit (no hardware needed)") args = ap.parse_args() if args.self_test: self_test() return if not args.port: ap.error("--port is required (or pass --self-test to just verify the parser)") csv_path = args.csv or datetime.now().strftime("fll_dac_%Y%m%d_%H%M%S.csv") stem = csv_path.rsplit(".", 1)[0] yfile_path = args.yfile or (stem + "_y.txt") events_path = stem + "_events.txt" run(args.port, args.baud, csv_path, args.nominal_hz, yfile_path, args.dac_full_scale_hz, events_path) if __name__ == "__main__": main()