#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ РАДИОЗАНАВЕС — живой визуализатор и детектор (пара к RFCurtain.ino, роль RX). Устройство глупое (льёт DATA,millis,seq,rssi,snr в консоль) — ВЕСЬ детектор здесь, чтобы дети крутили пороги и ВИДЕЛИ на графике, что меняется. Детектор (два признака, каждый — свой график): 1) ПРОСАДКА: быстрая медиана (окно ~0.5 c) упала ниже медленной базы на drop_db децибел. База — EMA с постоянной времени tau_base секунд, и она ЗАМОРАЖИВАЕТСЯ на время тревоги: иначе нарушитель, постояв в луче, станет «новой нормой» и тревога снимется сама (классическая ошибка охранных систем — покажите её детям, выключив заморозку). 2) КИПЕНИЕ: разброс (std) в быстром окне выше sigma_db — многолучёвка от движущегося тела заставляет сигнал «кипеть» даже без просадки. Плюс диагноз ОБРЫВ: пакетов нет дольше gap_s секунд — это НЕ нарушитель, это умер линк (питание TX, кто-то унёс антенну) — отдельный цвет. Кнопка на RX-плате шлёт MARK — «отметка правды» (судья жмёт в момент реального пересечения). На графике — вертикальная линия: честно считаем попадания, пропуски и ложные тревоги. Запуск (живой): python3 rf_curtain.py --port /dev/ttyUSB0 Повтор из лога: python3 rf_curtain.py --replay curtain_log.csv Пороги на лету: --drop-db 3.0 --sigma-db 2.5 --tau-base 60 Зависимости: pip install pyserial matplotlib numpy """ import argparse import collections import csv import sys import time import numpy as np import matplotlib import matplotlib.pyplot as plt from matplotlib.animation import FuncAnimation # ----------------------------- параметры CLI ----------------------------- ap = argparse.ArgumentParser(description="Радиозанавес: живой график + детектор") ap.add_argument("--port", help="последовательный порт RX-платы, напр. /dev/ttyUSB0 или COM5") ap.add_argument("--baud", type=int, default=115200) ap.add_argument("--replay", help="не слушать порт, а проиграть CSV-лог") ap.add_argument("--log", default=None, help="куда писать CSV-лог (по умолчанию curtain_YYYYmmdd_HHMMSS.csv)") ap.add_argument("--drop-db", type=float, default=3.0, help="порог просадки от базы, дБ (default 3.0)") ap.add_argument("--sigma-db", type=float, default=2.5, help="порог кипения (std быстрого окна), дБ (default 2.5)") ap.add_argument("--fast-win", type=float, default=0.5, help="быстрое окно, c (default 0.5)") ap.add_argument("--tau-base", type=float, default=60.0, help="постоянная времени базы, c (default 60)") ap.add_argument("--hold-s", type=float, default=1.0, help="удержание тревоги после ухода признаков, c") ap.add_argument("--gap-s", type=float, default=1.5, help="нет пакетов дольше -> ОБРЫВ линка, c") ap.add_argument("--warmup-s", type=float, default=8.0, help="прогрев базы: детектор молчит первые N c") ap.add_argument("--window-s", type=float, default=60.0, help="ширина живого графика, c") ap.add_argument("--no-freeze", action="store_true", help="НЕ замораживать базу во время тревоги (учебный анти-режим!)") args = ap.parse_args() if not args.port and not args.replay: ap.error("нужен --port (живой режим) или --replay файл.csv") # ----------------------------- состояние ----------------------------- t0 = time.monotonic() ts = collections.deque() # время, c (хостовое) rssi = collections.deque() # сырой RSSI fast_md = collections.deque() # быстрая медиана base = collections.deque() # медленная база sigma = collections.deque() # std быстрого окна marks = [] # отметки правды (MARK), времена events = [] # [t_start, t_end|None, вид('drop'/'boil')] gaps = [] # [t_start, t_end|None] обрывы линка fast_buf = collections.deque() # (t, rssi) быстрого окна baseline = None # EMA alarm_until = 0.0 alarm_active = False last_pkt_t = None gap_open = False n_pkts = 0 # ----------------------------- лог ----------------------------- log_path = args.log or time.strftime("curtain_%Y%m%d_%H%M%S.csv") log_f = None log_w = None if not args.replay: log_f = open(log_path, "w", newline="") log_w = csv.writer(log_f) log_w.writerow(["host_t", "kind", "seq", "rssi", "snr"]) # kind: DATA|MARK print(f"[лог] пишу в {log_path} (потом можно --replay)") # ----------------------------- источник строк ----------------------------- def line_source_serial(): import serial # pyserial ser = serial.Serial(args.port, args.baud, timeout=0.05) print(f"[порт] {args.port} @ {args.baud}. Жду строки DATA/MARK от RX-платы...") buf = b"" while True: chunk = ser.read(256) if chunk: buf += chunk while b"\n" in buf: line, buf = buf.split(b"\n", 1) yield line.decode(errors="replace").strip() else: yield None # нет данных прямо сейчас — отдаём управление анимации def line_source_replay(): # Проигрываем лог «как будто вживую» (в 1x времени по host_t) with open(args.replay) as f: rd = csv.DictReader(f) rows = list(rd) if not rows: sys.exit("[replay] пустой лог") t_first = float(rows[0]["host_t"]) t_start = time.monotonic() i = 0 print(f"[replay] {args.replay}: {len(rows)} строк") while i < len(rows): now_rel = time.monotonic() - t_start if float(rows[i]["host_t"]) - t_first <= now_rel: r = rows[i]; i += 1 if r["kind"] == "MARK": yield f"MARK,0" else: yield f"DATA,0,{r['seq']},{r['rssi']},{r['snr']}" else: yield None print("[replay] лог кончился — окно можно разглядывать, Ctrl+C для выхода") while True: yield None source = line_source_replay() if args.replay else line_source_serial() # ----------------------------- детектор ----------------------------- def feed_data(t, seq, r, snr): """Один принятый пакет: обновить окна, базу, решить про тревогу.""" global baseline, alarm_until, alarm_active, last_pkt_t, gap_open, n_pkts n_pkts += 1 last_pkt_t = t # обрыв закончился? if gap_open: gaps[-1][1] = t gap_open = False print(f"[{t:8.1f}c] ЛИНК ВОССТАНОВЛЕН") # быстрое окно fast_buf.append((t, r)) while fast_buf and t - fast_buf[0][0] > args.fast_win: fast_buf.popleft() vals = np.array([v for _, v in fast_buf]) fmd = float(np.median(vals)) fsd = float(np.std(vals)) if len(vals) >= 3 else 0.0 # медленная база (EMA). Заморозка во время тревоги — см. докстринг. if baseline is None: baseline = fmd freeze = alarm_active and not args.no_freeze if not freeze: dt = (t - ts[-1]) if ts else 0.07 a = min(1.0, dt / args.tau_base) baseline = baseline + a * (fmd - baseline) # признаки warm = t >= args.warmup_s drop = (baseline - fmd) >= args.drop_db boil = fsd >= args.sigma_db if warm and (drop or boil): if not alarm_active: kind = "drop" if drop else "boil" events.append([t, None, kind]) print(f"[{t:8.1f}c] ТРЕВОГА ({'просадка' if drop else 'кипение'}): " f"база={baseline:.1f} медиана={fmd:.1f} std={fsd:.1f}\a") alarm_active = True alarm_until = t + args.hold_s elif alarm_active and t > alarm_until: alarm_active = False events[-1][1] = t print(f"[{t:8.1f}c] отбой, длительность {t - events[-1][0]:.1f} c") ts.append(t); rssi.append(r); fast_md.append(fmd) base.append(baseline); sigma.append(fsd) # подрезаем живое окно (в replay храним всё) if not args.replay: while ts and t - ts[0] > args.window_s * 1.2: ts.popleft(); rssi.popleft(); fast_md.popleft() base.popleft(); sigma.popleft() def check_gap(now): """Тишина в эфире = обрыв линка, НЕ нарушитель. Отдельный диагноз.""" global gap_open if last_pkt_t is not None and not gap_open and now - last_pkt_t > args.gap_s: gaps.append([last_pkt_t, None]) gap_open = True print(f"[{now:8.1f}c] ОБРЫВ ЛИНКА: нет пакетов > {args.gap_s} c " f"(TX на паузе? питание? антенна?)\a") # ----------------------------- графика ----------------------------- plt.style.use("dark_background") fig, (ax1, ax2, ax3) = plt.subplots( 3, 1, figsize=(12, 8), sharex=True, gridspec_kw={"height_ratios": [3, 1.4, 1.4]}) fig.canvas.manager.set_window_title("РАДИОЗАНАВЕС — живой эфир") fig.suptitle("Радиозанавес: тень человека в первой зоне Френеля", fontsize=13, fontweight="bold") ln_raw, = ax1.plot([], [], ".", ms=3, color="#4a5a6a", alpha=0.6, label="сырой RSSI") ln_fast, = ax1.plot([], [], "-", lw=2, color="#2dd4bf", label="быстрая медиана (0.5 c)") ln_base, = ax1.plot([], [], "--", lw=1.6, color="#f5c542", label="медленная база") ax1.set_ylabel("RSSI, дБм") ax1.legend(loc="lower left", fontsize=9, framealpha=0.3) ax1.grid(alpha=0.2) ln_dev, = ax2.plot([], [], "-", lw=1.6, color="#60a5fa") ax2.axhline(0, color="#666", lw=0.8) thr_dev = ax2.axhline(-args.drop_db, color="#ef4444", ls=":", lw=1.4) ax2.set_ylabel("медиана − база, дБ") ax2.grid(alpha=0.2) ln_sig, = ax3.plot([], [], "-", lw=1.6, color="#c084fc") thr_sig = ax3.axhline(args.sigma_db, color="#ef4444", ls=":", lw=1.4) ax3.set_ylabel("кипение (std), дБ") ax3.set_xlabel("время, c") ax3.grid(alpha=0.2) status_txt = ax1.text(0.99, 0.95, "", transform=ax1.transAxes, ha="right", va="top", fontsize=12, fontweight="bold") spans = [] # закраски событий/обрывов (перерисовываем целиком) mark_lines = [] def update(_frame): # вычитываем всё, что накопилось (порт быстрее кадра анимации) for _ in range(400): line = next(source) if line is None: break now = time.monotonic() - t0 if line.startswith("DATA,"): try: _, _ms, seq, r, snr = line.split(",") feed_data(now, int(seq), float(r), float(snr)) if log_w: log_w.writerow([f"{now:.3f}", "DATA", seq, r, snr]) except ValueError: pass elif line.startswith("MARK"): marks.append(now) print(f"[{now:8.1f}c] === ОТМЕТКА ПРАВДЫ (судья) ===") if log_w: log_w.writerow([f"{now:.3f}", "MARK", "", "", ""]) elif line.startswith(("===", "[")): print(f"[плата] {line}") now = time.monotonic() - t0 check_gap(now) if not ts: return [] x = np.array(ts) ln_raw.set_data(x, np.array(rssi)) ln_fast.set_data(x, np.array(fast_md)) ln_base.set_data(x, np.array(base)) ln_dev.set_data(x, np.array(fast_md) - np.array(base)) ln_sig.set_data(x, np.array(sigma)) xmax = x[-1] xmin = 0 if args.replay else max(0, xmax - args.window_s) for ax in (ax1, ax2, ax3): ax.set_xlim(xmin, max(xmax, xmin + 5)) ax1.relim(); ax1.autoscale_view(scalex=False) ax2.relim(); ax2.autoscale_view(scalex=False) ax3.relim(); ax3.autoscale_view(scalex=False) # события/обрывы/метки — перерисовать for s in spans: s.remove() spans.clear() for t_a, t_b, kind in events: c = "#ef4444" if kind == "drop" else "#f97316" spans.append(ax1.axvspan(t_a, t_b or xmax, color=c, alpha=0.18)) for t_a, t_b in gaps: spans.append(ax1.axvspan(t_a, t_b or xmax, color="#3b82f6", alpha=0.15)) for ml in mark_lines: ml.remove() mark_lines.clear() for tm in marks: for ax in (ax1, ax2, ax3): mark_lines.append(ax.axvline(tm, color="#ffffff", ls="--", lw=1.0, alpha=0.7)) if gap_open: status_txt.set_text("ОБРЫВ ЛИНКА"); status_txt.set_color("#60a5fa") elif alarm_active: status_txt.set_text("ТРЕВОГА"); status_txt.set_color("#ef4444") elif now < args.warmup_s: status_txt.set_text("прогрев базы..."); status_txt.set_color("#f5c542") else: status_txt.set_text("охраняется"); status_txt.set_color("#22c55e") return [] ani = FuncAnimation(fig, update, interval=100, cache_frame_data=False) print("\nПодсказки для урока:") print(f" порог просадки {args.drop_db} дБ, порог кипения {args.sigma_db} дБ " f"(крутить ключами --drop-db / --sigma-db)") print(" красная заливка = тревога-просадка, оранжевая = тревога-кипение,") print(" синяя = обрыв линка (это ДРУГОЙ диагноз!), белый пунктир = отметка правды.") if args.no_freeze: print(" ВНИМАНИЕ: --no-freeze — база ползёт за нарушителем (учебный анти-режим).") try: plt.show() finally: if log_f: log_f.close() print(f"[лог] сохранён: {log_path}") # мини-итог для разбора с детьми n_ev = len(events) print(f"Итог сессии: пакетов={n_pkts}, тревог={n_ev}, отметок правды={len(marks)}, обрывов={len(gaps)}") if marks and n_ev: hits = sum(1 for tm in marks if any(a - 1.0 <= tm <= (b or tm) + 1.0 for a, b, _ in events)) print(f" попаданий (тревога рядом с отметкой ±1 c): {hits}/{len(marks)}, " f"остальные тревоги — кандидаты в ложные. Это и есть ваш ROC-опыт.")