""" PROJECT SLOW-SWEEP — Live monitor with beep counting Watches: encoder position, IR frames, and mic beep patterns simultaneously. Run: python3 monitor.py Tail: tail -f monitor.log """ import serial, time, subprocess, threading, sys import numpy as np LOG = "/home/person/Desktop/Turntable_Hacking/monitor.log" WROOM_PORT = "/dev/ttyACM0" MIC_DEVICE = "hw:3,0" MIC_RATE = 44100 # ── Logging ────────────────────────────────────────────────────────────────── def log(msg, tag=""): ts = time.strftime("%H:%M:%S.") + f"{int(time.time()*10)%10}" prefix = f"[{tag}]" if tag else "" line = f"[{ts}] {prefix} {msg}" print(line, flush=True) with open(LOG, "a") as f: f.write(line + "\n") # ── Beep counter ───────────────────────────────────────────────────────────── class BeepCounter: """ Continuously records audio and counts distinct beeps. A beep = RMS above threshold for >40ms. A new beep = gap of >120ms silence between spikes. Reports beep count whenever a pattern completes (silence after beeps). """ CHUNK_MS = 20 # analyze in 20ms windows MIN_BEEP_MS = 40 # minimum duration to count as a beep MIN_GAP_MS = 120 # silence gap to separate beeps REPORT_MS = 400 # silence after last beep before reporting def __init__(self, threshold=500): self.threshold = threshold self._stop = threading.Event() self._thread = None self.ambient = 0.0 def calibrate(self, seconds=2.0): """Measure ambient noise floor.""" log(f"Calibrating mic for {seconds}s...", "MIC") proc = subprocess.Popen( ['arecord', '-D', MIC_DEVICE, '-f', 'S16_LE', '-r', str(MIC_RATE), '-c', '1', '-d', str(int(seconds)+1)], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) data = proc.stdout.read() proc.kill() if data: samples = np.frombuffer(data, dtype=np.int16).astype(np.float32) self.ambient = float(np.sqrt(np.mean(samples**2))) self.threshold = max(self.ambient * 1.3, 50) log(f"Ambient RMS={self.ambient:.1f} Beep threshold={self.threshold:.1f}", "MIC") def start(self): self._stop.clear() self._thread = threading.Thread(target=self._run, daemon=True) self._thread.start() def stop(self): self._stop.set() def _run(self): chunk_samples = int(MIC_RATE * self.CHUNK_MS / 1000) # mono chunk_bytes = chunk_samples * 2 # 16-bit proc = subprocess.Popen( ['arecord', '-D', MIC_DEVICE, '-f', 'S16_LE', '-r', str(MIC_RATE), '-c', '1'], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) beep_count = 0 in_beep = False beep_ms = 0 gap_ms = 0 while not self._stop.is_set(): chunk = proc.stdout.read(chunk_bytes) if not chunk: break samples = np.frombuffer(chunk, dtype=np.int16).astype(np.float32) rms = float(np.sqrt(np.mean(samples**2))) loud = rms > self.threshold if loud: if not in_beep: in_beep = True beep_ms = 0 gap_ms = 0 else: beep_ms += self.CHUNK_MS gap_ms = 0 else: if in_beep: gap_ms += self.CHUNK_MS if gap_ms >= self.MIN_GAP_MS: # Beep ended if beep_ms >= self.MIN_BEEP_MS: beep_count += 1 in_beep = False beep_ms = 0 else: # In silence if beep_count > 0: gap_ms += self.CHUNK_MS if gap_ms >= self.REPORT_MS: log(f"BEEP x{beep_count}", "MIC") beep_count = 0 gap_ms = 0 proc.kill() if beep_count > 0: log(f"BEEP x{beep_count} (final)", "MIC") # ── Serial monitor ──────────────────────────────────────────────────────────── def serial_monitor(wroom, enc_ref): """Read WROOM serial and log IR frames, encoder changes, and alive ticks.""" last_enc = None last_ir_time = 0 while True: try: line = wroom.readline().decode(errors="replace").strip() except: break if not line: continue if "[ENC]" in line: # Only log every 10 counts to avoid spam, but always log direction changes try: counts = int(line.split("counts=")[1].split()[0]) deg_abs = round((counts - enc_ref) * 0.15, 1) if last_enc is None or abs(counts - last_enc) >= 10: log(f"pos={counts} ({deg_abs:+.1f}deg from home)", "ENC") last_enc = counts except: pass elif "[IR]" in line and "NEC decoded" in line: try: val = line.split("value=")[1].strip() cmd = (int(val, 16) >> 16) & 0xFF known = { 0x40:"step", 0x41:"ccw", 0x42:"startstop", 0x43:"slowdown", 0x44:"cw", 0x45:"continuous", 0x02:"45deg", 0x07:"180deg", 0x0A:"speedup", 0x11:"swing", 0x1B:"90deg", } name = known.get(cmd, f"0x{cmd:02X}") now = time.time() if now - last_ir_time > 0.5: # debounce repeats log(f"cmd={name} (0x{cmd:02X})", "IR") last_ir_time = now except: pass elif "[ALIVE]" in line: try: enc = int(line.split("enc=")[1].split()[0]) deg_abs = round((enc - enc_ref) * 0.15, 1) log(f"heartbeat enc={enc} ({deg_abs:+.1f}deg from home)", "TICK") if last_enc is None: last_enc = enc except: pass # ── Main ────────────────────────────────────────────────────────────────────── log("=== MONITOR START ===") log("Tail this log: tail -f /home/person/Desktop/Turntable_Hacking/monitor.log") wroom = serial.Serial(WROOM_PORT, 115200, timeout=1) time.sleep(2) wroom.reset_input_buffer() # Get initial encoder position as home reference log("Reading home position...", "SETUP") enc_home = None deadline = time.time() + 5 while time.time() < deadline: line = wroom.readline().decode(errors="replace").strip() if "[ALIVE]" in line and "enc=" in line: try: enc_home = int(line.split("enc=")[1].split()[0]) break except: pass if enc_home is None: enc_home = 0 log(f"Home encoder position = {enc_home}", "SETUP") # Calibrate mic mic = BeepCounter() mic.calibrate(seconds=2.0) mic.start() log("", "") log("=== READY — start your test sequence ===") log("SEQUENCE: reboot → mark home → 180deg x2 CW → switch CCW → 180deg x2 CCW") log("") # Run serial monitor (blocks until serial dies) try: serial_monitor(wroom, enc_home) except KeyboardInterrupt: pass finally: mic.stop() log("=== MONITOR STOPPED ===")