monitor.py

#!/usr/bin/env python3
"""Firefly Synchronization Monitor — TUI for iCEstick serial telemetry.

Packet format (16 bytes, ~11 Hz):
  0x55  sync
  5 × { phase_hi[7:0], phase_lo[7:0], bright[7:0] }

Phase[15:0] is the top 16 bits of the 32-bit DDS accumulator.
Flash zone: phase >= 0xE000 (top 3 bits = 111).
"""
import argparse, collections, sys, time

SYNC = 0x55
FLY_BYTES = 3
N_FLIES = 5
PKT_LEN = 1 + N_FLIES * FLY_BYTES  # 16
NOTES = ['A4 ', 'C5 ', 'D5 ', 'E5 ', 'G5 ']
PAN   = ['◀◀', '◀ ', '◆ ', ' ▶', '▶▶']  # stereo pan labels
BAR_W = 40
RING_W = 64
SPARK_W = 40
SPREAD_SYNC = 0.04  # threshold for "synced"

# ANSI helpers
CSI    = '\033['
CLEAR  = f'{CSI}2J{CSI}H'
HIDE   = f'{CSI}?25l'
SHOW   = f'{CSI}?25h'
RESET  = f'{CSI}0m'
BOLD   = f'{CSI}1m'
DIM    = f'{CSI}2m'
FG_Y   = f'{CSI}33m'
FG_R   = f'{CSI}91m'
FG_G   = f'{CSI}92m'
FG_C   = f'{CSI}96m'
BG_R   = f'{CSI}41m'

FIRE   = [f'{CSI}38;5;{c}m' for c in (208, 214, 220, 226, 228)]
FIRE_DIM = [f'{CSI}38;5;{c}m' for c in (130, 136, 142, 178, 180)]
FG_FLASH = f'{CSI}38;5;230m'  # bright cream for flash zone glow
SPARK_CHARS = ' ▁▂▃▄▅▆▇█'

def goto(r, c=1):
    return f'{CSI}{r};{c}H'

def phase_bar(phase16, color, flash):
    """BAR_W-char ring with colored position dot and flash zone glow."""
    pos = int(phase16 / 0x10000 * BAR_W)
    fz_start = int(BAR_W * 7 / 8)
    chars = []
    for i in range(BAR_W):
        if i == pos % BAR_W:
            chars.append(f'{color}{BOLD}●{RESET}')
        elif i >= fz_start:
            if flash:
                chars.append(f'{color}░{RESET}')
            else:
                chars.append(f'{DIM}░{RESET}')
        else:
            chars.append(f'{DIM}·{RESET}')
    return ''.join(chars)

def constellation(flies):
    """Single wide ring showing all 5 flies together."""
    fz_start = int(RING_W * 7 / 8)
    slots = [None] * RING_W
    # place flies (later ones overwrite — rare at this resolution)
    for i, (ph, _br) in enumerate(flies):
        p = int(ph / 0x10000 * RING_W) % RING_W
        slots[p] = i
    chars = []
    for j in range(RING_W):
        if slots[j] is not None:
            chars.append(f'{FIRE[slots[j]]}{BOLD}●{RESET}')
        elif j >= fz_start:
            chars.append(f'{DIM}░{RESET}')
        else:
            chars.append(f'{DIM}·{RESET}')
    return ''.join(chars)

def sparkline(history):
    """Tiny ASCII sparkline of spread values (0..1 → char)."""
    if not history:
        return ''
    mx = max(history) or 1
    out = []
    for v in history:
        idx = min(int(v / mx * (len(SPARK_CHARS) - 1)), len(SPARK_CHARS) - 1)
        if v < SPREAD_SYNC:
            out.append(f'{FG_G}{SPARK_CHARS[idx]}{RESET}')
        else:
            out.append(f'{FG_C}{SPARK_CHARS[idx]}{RESET}')
    return ''.join(out)

def fmt_dur(secs):
    """Format seconds as M:SS or H:MM:SS."""
    s = int(secs)
    if s < 3600:
        return f'{s // 60}:{s % 60:02d}'
    return f'{s // 3600}:{(s % 3600) // 60:02d}:{s % 60:02d}'

def main():
    ap = argparse.ArgumentParser(description='Firefly sync monitor')
    ap.add_argument('-p', '--port', default='/dev/ttyUSB1')
    ap.add_argument('-b', '--baud', type=int, default=115200)
    args = ap.parse_args()

    try:
        import serial
    except ImportError:
        print('pip install pyserial', file=sys.stderr)
        sys.exit(1)

    ser = serial.Serial(args.port, args.baud, timeout=0.5)
    ser.reset_input_buffer()

    pkts = 0
    t0 = time.monotonic()
    sync_since = None       # timestamp when sync started (None = not synced)
    last_sync_dur = None    # duration of last sync period
    spread_hist = collections.deque(maxlen=SPARK_W)

    print(HIDE + CLEAR, end='')
    print(f'{BOLD}  🔥 Firefly Synchronization Monitor{RESET}')
    print(f'  {DIM}Port: {args.port} @ {args.baud}{RESET}')
    print(f'  {DIM}{"─" * 68}{RESET}')

    try:
        while True:
            # ── sync to packet ──
            while True:
                b = ser.read(1)
                if not b:
                    continue
                if b[0] == SYNC:
                    break

            data = ser.read(N_FLIES * FLY_BYTES)
            if len(data) < N_FLIES * FLY_BYTES:
                continue

            pkts += 1
            flies = []
            for i in range(N_FLIES):
                o = i * FLY_BYTES
                ph = (data[o] << 8) | data[o + 1]
                br = data[o + 2]
                flies.append((ph, br))

            # ── phase spread (handle wraparound) ──
            phases = sorted(f[0] for f in flies)
            gaps = [(phases[(i+1) % N_FLIES] - phases[i]) % 0x10000
                    for i in range(N_FLIES)]
            spread = 1.0 - max(gaps) / 0x10000
            synced = spread < SPREAD_SYNC
            spread_hist.append(spread)

            # ── sync timer ──
            now = time.monotonic()
            if synced:
                if sync_since is None:
                    sync_since = now
            else:
                if sync_since is not None:
                    last_sync_dur = now - sync_since
                    sync_since = None

            # ── render per-fly rows ──
            out = goto(5)
            for i, (ph, br) in enumerate(flies):
                pct = ph / 0xFFFF * 100
                flash = ph >= 0xE000
                col = FIRE[i]
                bbar_n = int(br / 255 * 16) if br else 0
                bbar = '█' * bbar_n + ' ' * (16 - bbar_n)
                ind = f'{FG_Y}⚡{RESET}' if flash else '  '
                pbar = phase_bar(ph, col, flash)
                out += (f'  {col}{NOTES[i]}{RESET}'
                        f'{DIM}{PAN[i]}{RESET} '
                        f'{DIM}ph{RESET} {pct:5.1f}% '
                        f'{DIM}br{RESET} {br:3d} {ind} '
                        f'{col}{bbar}{RESET} '
                        f'{pbar}\n')

            # ── constellation ring ──
            out += f'\n  {DIM}ring{RESET}  {constellation(flies)}\n'

            # ── sync status ──
            out += f'\n  Spread: {spread*100:5.2f}%  '
            if synced:
                dur = now - sync_since if sync_since else 0
                out += f'{FG_G}{BOLD}● SYNCED{RESET} {FG_G}{fmt_dur(dur)}{RESET}  '
                out += f'{FG_G}♪ melody active{RESET}     '
            else:
                out += f'{FG_C}○ converging...{RESET}  '
                if last_sync_dur is not None:
                    out += f'{DIM}last sync held {fmt_dur(last_sync_dur)}{RESET}'
                else:
                    out += f'{DIM}waiting for first sync{RESET}'

            # ── sparkline ──
            out += f'\n  {DIM}spread{RESET} {sparkline(spread_hist)}'

            # ── footer ──
            elapsed = now - t0
            rate = pkts / elapsed if elapsed > 0 else 0
            out += f'\n  {DIM}{rate:.1f} pkt/s    Ctrl-C to quit{RESET}'
            print(out, end='', flush=True)

    except KeyboardInterrupt:
        pass
    finally:
        print(SHOW + RESET + '\n')

if __name__ == '__main__':
    main()