#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ РАДИОЗАНАВЕС — СИМУЛЯТОР для класса (без железа, без пары TX/RX). Зачем нужен рядом с боевым rf_curtain.py: тот детектор требует живого линка (две платы E77) или записанного лога. Этот стенд СИНТЕЗИРУЕТ поток RSSI — человек «идёт» по сцене, пересекает луч, RSSI проседает и «кипит» ровно по физике из Теория_RFCurtain.md (раздел 3), — и прогоняет ТОТ ЖЕ детектор с двумя признаками. Ученик крутит пороги ползунками и сразу видит, что стало с попаданиями и ложными тревогами. Урок «настрой детектор» (Writeup, опыт В) можно провести за партой до выхода во двор. Физика синтеза (всё из теории, раздел 3): * зона Френеля: r₁ = √(λ·d1·d2/d); человек перекрывает долю площади зоны в точке пересечения → просадка тем глубже, чем ближе к середине луча; * экранировка телом: до ~8–10 дБ поглощения на 869 МГц; * многолучёвка: движущееся тело перемешивает отражения → растёт РАЗБРОС (std) RSSI — второй признак «кипение», даже без большой просадки; * дрейф фона: медленный синус + шум, чтобы показать, зачем нужна ПЛАВАЮЩАЯ база, а не абсолютный порог. Детектор — копия правила из rf_curtain.py: быстрая медиана (окно 0.5 c) против медленной базы (EMA), тревога по просадке ИЛИ по кипению, база ЗАМОРАЖИВАЕТСЯ на время тревоги (кнопка «Заморозка базы» выключает заморозку — учебный анти-режим: нарушитель, постояв в луче, становится «новой нормой»). Запуск: uv run rf_curtain_sim.py Без uv: pip install numpy matplotlib && python rf_curtain_sim.py """ # /// script # requires-python = ">=3.11" # dependencies = ["numpy>=1.26", "matplotlib>=3.8"] # /// import numpy as np import matplotlib.pyplot as plt from matplotlib.widgets import Slider, Button # --- Параметры линка (1:1 с RFCurtain.ino / Теория раздел 3) --- FREQ_MHZ = 869.400 C = 299_792_458.0 LAMBDA_M = C / (FREQ_MHZ * 1e6) # ~0.345 м на 869 МГц LINK_LEN_M = 10.0 # расстояние TX↔RX RSSI_BASE_DBM = -45.0 # «жирный» линк вблизи (стол/короткая дистанция) BODY_ATTEN_DB = 9.0 # поглощение телом при полном перекрытии зоны PERSON_RADIO_M = 0.5 # «радиотень» человека RATE_HZ = 15.0 # частота пакетов занавеса DUR_S = 40.0 # длительность синтезируемой сессии FAST_WIN_S = 0.5 TAU_BASE_S = 8.0 # для класса база быстрее (короче сессия) HOLD_S = 1.0 WARMUP_S = 4.0 def fresnel_radius_m(d1, d2): """Радиус первой зоны Френеля в точке, делящей линк на d1 и d2 (Теория 3).""" d = d1 + d2 if d <= 0: return 0.0 return np.sqrt(LAMBDA_M * d1 * d2 / d) def synth_trace(seed, crossings): """ Синтезирует (t, rssi) и список моментов пересечения (ground truth). crossings: список (t_center, где_вдоль_луча 0..1, скорость 'шаг'/'бег'). """ rng = np.random.default_rng(seed) n = int(DUR_S * RATE_HZ) t = np.arange(n) / RATE_HZ # фон: медленный дрейф (температура/фон) + белый шум измерения drift = 1.5 * np.sin(2 * np.pi * t / 25.0) rssi = RSSI_BASE_DBM + drift + rng.normal(0, 0.8, n) truth = [] for (tc, pos, speed) in crossings: truth.append(tc) # ширина «ямы» во времени: бег короче шага width = 0.25 if speed == "бег" else 0.6 # Глубина просадки. Основа — экранировка телом (BODY_ATTEN_DB). Она # максимальна в СЕРЕДИНЕ луча, где радиус зоны Френеля наибольший # (Теория 3): тело закрывает бОльшую долю «чувствительного объёма». # Ближе к антеннам зона узкая — тело проскакивает край, тень мельче. d1 = max(pos, 0.02) * LINK_LEN_M d2 = max(1.0 - pos, 0.02) * LINK_LEN_M r1 = fresnel_radius_m(d1, d2) r1_mid = fresnel_radius_m(LINK_LEN_M / 2, LINK_LEN_M / 2) centrality = r1 / r1_mid # 1.0 в центре, спадает к краям depth = BODY_ATTEN_DB * centrality env = np.exp(-0.5 * ((t - tc) / width) ** 2) # гауссова «яма» rssi -= depth * env # многолучёвка: во время прохода растёт РАЗБРОС (кипение) boil = rng.normal(0, 1.0, n) * (2.5 * env) rssi += boil return t, rssi, truth def run_detector(t, rssi, drop_db, sigma_db, freeze_on): """Копия правила из rf_curtain.py. Возвращает (fast_md, base, sigma, events).""" fast_md = np.zeros_like(rssi) base_arr = np.zeros_like(rssi) sig_arr = np.zeros_like(rssi) baseline = None alarm_active = False alarm_until = 0.0 events = [] # [t_start, t_end|None, kind] from collections import deque fast_buf = deque() for i, (ti, ri) in enumerate(zip(t, rssi)): fast_buf.append((ti, ri)) while fast_buf and ti - fast_buf[0][0] > FAST_WIN_S: 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 if baseline is None: baseline = fmd freeze = alarm_active and freeze_on if not freeze: dt = (ti - t[i - 1]) if i > 0 else 1.0 / RATE_HZ a = min(1.0, dt / TAU_BASE_S) baseline += a * (fmd - baseline) warm = ti >= WARMUP_S drop = (baseline - fmd) >= drop_db boil = fsd >= sigma_db if warm and (drop or boil): if not alarm_active: events.append([ti, None, "drop" if drop else "boil"]) alarm_active = True alarm_until = ti + HOLD_S elif alarm_active and ti > alarm_until: alarm_active = False events[-1][1] = ti fast_md[i] = fmd base_arr[i] = baseline sig_arr[i] = fsd if events and events[-1][1] is None: events[-1][1] = t[-1] return fast_md, base_arr, sig_arr, events def score(events, truth): """ROC-подсчёт: попадания (тревога рядом с пересечением ±1 c) и ложные.""" hits = 0 matched = [False] * len(events) for tm in truth: for k, (a, b, _) in enumerate(events): if a - 1.0 <= tm <= (b or tm) + 1.0: hits += 1 matched[k] = True break false_alarms = sum(1 for k, m in enumerate(matched) if not m) misses = len(truth) - hits return hits, misses, false_alarms def main(): # Сценарий: пересечение шагом в середине, пробежка сбоку (провокация), # пересечение бегом ближе к RX. Судья «отмечает» только реальные пересечения. crossings = [(10.0, 0.5, "шаг"), (20.0, 0.5, "бег"), (30.0, 0.75, "шаг")] provocation_only = 25.0 # «кипение» рядом с лучом без пересечения — НЕ ground truth st = {"seed": 1, "drop": 3.0, "sigma": 2.5, "freeze": True} fig = plt.figure(figsize=(12, 8)) plt.style.use("default") fig.suptitle("Радиозанавес — СИМУЛЯТОР: тень человека в зоне Френеля + детектор", fontsize=13, fontweight="bold") ax1 = fig.add_axes([0.08, 0.62, 0.88, 0.28]) ax2 = fig.add_axes([0.08, 0.44, 0.88, 0.14]) ax3 = fig.add_axes([0.08, 0.26, 0.88, 0.14]) def redraw(_=None): t, rssi, truth = synth_trace(st["seed"], crossings) # добавим «провокацию» (кипение без просадки) — не в truth rng = np.random.default_rng(st["seed"] + 100) env = np.exp(-0.5 * ((t - provocation_only) / 0.7) ** 2) rssi = rssi + rng.normal(0, 1.0, len(t)) * (2.8 * env) fmd, base, sig, events = run_detector(t, rssi, st["drop"], st["sigma"], st["freeze"]) hits, misses, fa = score(events, truth) for ax in (ax1, ax2, ax3): ax.clear() # --- верхний: сырьё + два сглаживания --- ax1.plot(t, rssi, ".", ms=2.5, color="#9aa5b1", alpha=0.6, label="сырой RSSI") ax1.plot(t, fmd, "-", lw=2, color="#0d9488", label="быстрая медиана (0.5 c)") ax1.plot(t, base, "--", lw=1.6, color="#d97706", label="медленная база (EMA)") for a, b, kind in events: ax1.axvspan(a, b, color="#ef4444" if kind == "drop" else "#f97316", alpha=0.18) for tm in truth: ax1.axvline(tm, color="#111", ls="--", lw=1.0, alpha=0.7) ax1.set_ylabel("RSSI, дБм") ax1.legend(loc="lower left", fontsize=8, framealpha=0.4) ax1.grid(alpha=0.25) fr = "заморозка ВКЛ" if st["freeze"] else "заморозка ВЫКЛ (анти-режим!)" ax1.set_title(f"Попаданий: {hits}/{len(truth)} | пропусков: {misses} | " f"ложных: {fa} ({fr})", fontsize=10) # --- средний: признак 1, просадка --- ax2.plot(t, fmd - base, "-", lw=1.5, color="#2563eb") ax2.axhline(-st["drop"], color="#ef4444", ls=":", lw=1.4) ax2.axhline(0, color="#999", lw=0.7) ax2.set_ylabel("медиана−база, дБ") ax2.grid(alpha=0.25) # --- нижний: признак 2, кипение --- ax3.plot(t, sig, "-", lw=1.5, color="#9333ea") ax3.axhline(st["sigma"], color="#ef4444", ls=":", lw=1.4) ax3.set_ylabel("кипение std, дБ") ax3.set_xlabel("время, c") ax3.grid(alpha=0.25) for ax in (ax1, ax2, ax3): ax.set_xlim(0, DUR_S) fig.canvas.draw_idle() ax_drop = fig.add_axes([0.14, 0.15, 0.5, 0.03]) s_drop = Slider(ax_drop, "Порог просадки, дБ", 0.5, 8.0, valinit=st["drop"], valstep=0.5) ax_sig = fig.add_axes([0.14, 0.10, 0.5, 0.03]) s_sig = Slider(ax_sig, "Порог кипения σ, дБ", 0.5, 6.0, valinit=st["sigma"], valstep=0.5) def on_drop(v): st["drop"] = v; redraw() def on_sig(v): st["sigma"] = v; redraw() s_drop.on_changed(on_drop) s_sig.on_changed(on_sig) ax_fr = fig.add_axes([0.72, 0.13, 0.22, 0.05]) b_fr = Button(ax_fr, "Заморозка базы: ВКЛ/ВЫКЛ") def toggle_freeze(_): st["freeze"] = not st["freeze"]; redraw() b_fr.on_clicked(toggle_freeze) ax_new = fig.add_axes([0.72, 0.06, 0.22, 0.05]) b_new = Button(ax_new, "Новая сессия (шум)") def reseed(_): st["seed"] += 1; redraw() b_new.on_clicked(reseed) redraw() plt.show() if __name__ == "__main__": main()