#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ POWER SENTRY — СИМУЛЯТОР ГОРОДСКОЙ СЕТИ (проект «Инженеры будущего»). Тезис проекта: контроль протечек в городском ЖКХ масштабируется на ВЕСЬ город не потому, что LoRa «дальнобойная», а потому, что LoRaWAN даёт крайнюю ЭНЕРГОЭФФЕКТИВНОСТЬ — узел спит годами и выходит в эфир только по делу. Именно редкость уплинков (событийная модель + send-on-change) одновременно решает две задачи: годы жизни от батарейки И десятки тысяч узлов на один шлюз. Этот стенд делает тезис измеримым. Он моделирует город из N узлов-датчиков протечки и M шлюзов и считает по РЕАЛЬНЫМ формулам LoRaWAN: 1. ЭФИРНОЕ ВРЕМЯ (airtime) уплинка по SF/BW (формула Semtech: preamble + payload-символы), отсюда нагрузка на канал. 2. ЁМКОСТЬ ШЛЮЗА как ALOHA-канал: при суммарной интенсивности G Эрланг доля успешно принятых кадров ≈ e^(−2G) (чистая ALOHA). Так видно, при какой частоте уплинков сеть «захлёбывается». 3. СРОК ЖИЗНИ узла от пары AA по бюджету заряда (ток сна Stop2 + активность на замер/TX) — та же арифметика, что в README проекта. Ключевая демонстрация (двигайте ползунок «уплинков в сутки на узел»): * событийная модель ЖКХ (~1-2 уплинка/сутки на узел): десятки тысяч узлов на шлюз, PDR ≈ 100%, годы от батарейки; * «наивная» периодическая (уплинк раз в минуту): та же сеть держит <500 узлов, PDR обваливается, батарея тает — ровно то, чего проект избегает. Значения airtime/энергии согласованы с PowerSentry.ino (SF10/BW125, payload 10 Б, TX ~45 мА, Stop2 ~2 мкА). Числа capacity подтверждаются литературой по LoRaWAN (сенсор раз в сутки → 50k+ узлов/шлюз; раз в минуту → <500). Запуск: uv run sentry_city_sim.py Без uv: pip install numpy matplotlib && python sentry_city_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 # --- Радио/энергия: 1:1 с PowerSentry.ino --- PAYLOAD_BYTES = 10 # SentryPayload_t BW_KHZ = 125.0 PREAMBLE = 8 CR = 1 # coding rate 4/(4+CR), CR=1 => 4/5 N_CHANNELS = 8 # типовой шлюз SX1301: 8 × 125 кГц каналов DUTY_CYCLE = 0.01 # 1% в канале 868.1 (EU868) # Энергобюджет узла (из README, статья «энергобюджет») I_SLEEP_UA = 2.0 # Stop2, кристалл I_TX_MA = 45.0 # пик TX T_TX_ACTIVE_EXTRA_S = 0.05 # накладные до/после эфира (пробуждение радио) I_MEAS_MA = 1.0 # ток замера датчика T_MEAS_S = 0.2 BATTERY_MAH = 2500.0 # пара AA def airtime_s(sf, payload=PAYLOAD_BYTES, bw_khz=BW_KHZ): """Время кадра в эфире по формуле Semtech (LoRa Modem Designer's Guide).""" bw = bw_khz * 1000.0 t_sym = (2 ** sf) / bw # low-data-rate optimize при SF11/12 на BW125 de = 1 if (bw_khz == 125.0 and sf >= 11) else 0 n_preamble = (PREAMBLE + 4.25) * t_sym # header enabled (H=0), CRC on (уже в LoRa PHY) payload_sym = 8 + max( np.ceil((8 * payload - 4 * sf + 28 + 16) / (4 * (sf - 2 * de))) * (CR + 4), 0, ) return n_preamble + payload_sym * t_sym def node_life_years(uplinks_per_day, sf): """Срок жизни узла от пары AA по бюджету заряда.""" ta = airtime_s(sf) # заряд на один уплинк (мА·ч): (airtime+накладные)*I_TX + замер q_tx = (ta + T_TX_ACTIVE_EXTRA_S) / 3600.0 * I_TX_MA q_meas = T_MEAS_S / 3600.0 * I_MEAS_MA q_uplink = q_tx + q_meas # мА·ч на событие q_sleep_day = 24.0 * (I_SLEEP_UA / 1000.0) # мА·ч/сутки сна q_day = q_sleep_day + uplinks_per_day * q_uplink return BATTERY_MAH / q_day / 365.0, q_day, q_sleep_day, uplinks_per_day * q_uplink def aloha_pdr(n_nodes, uplinks_per_day, sf): """ Доля успешно доставленных кадров как чистая ALOHA. Интенсивность G (Эрланг) на КАНАЛ = средн. число кадров за уязвимое окно 2*T. Кадры равномерно раскиданы по N_CHANNELS каналам. """ ta = airtime_s(sf) frames_per_sec = n_nodes * uplinks_per_day / 86400.0 per_channel = frames_per_sec / N_CHANNELS G = per_channel * ta # Эрланг на канал pdr = np.exp(-2.0 * G) # pure ALOHA S = G·e^(-2G); PDR=e^(-2G) return pdr, G def duty_max_uplinks_per_day(sf): """Сколько уплинков/сутки на узел позволяет 1% duty (для справки).""" ta = airtime_s(sf) return DUTY_CYCLE * 86400.0 / ta def main(): st = {"nodes": 20000, "uplinks": 2.0, "sf": 10} fig = plt.figure(figsize=(13, 8)) fig.suptitle("Power Sentry — городская сеть ЖКХ: почему LoRaWAN масштабируется", fontsize=14, fontweight="bold") ax_cap = fig.add_axes([0.07, 0.42, 0.40, 0.46]) # PDR vs узлы ax_life = fig.add_axes([0.57, 0.42, 0.40, 0.46]) # срок жизни vs частота ax_txt = fig.add_axes([0.07, 0.20, 0.90, 0.14]) ax_txt.axis("off") def redraw(_=None): sf = st["sf"] ta = airtime_s(sf) # --- ЛЕВО: PDR от числа узлов при текущей частоте уплинков --- ax_cap.clear() nn = np.logspace(2, 5.5, 200) pdr = np.array([aloha_pdr(n, st["uplinks"], sf)[0] for n in nn]) ax_cap.plot(nn, pdr * 100, color="#2563eb", lw=2) ax_cap.axhline(95, color="#16a34a", ls="--", lw=1) ax_cap.text(1.2e2, 96, "95% PDR (рабочий порог)", color="#16a34a", fontsize=8) # текущая точка cur_pdr, G = aloha_pdr(st["nodes"], st["uplinks"], sf) ax_cap.plot(st["nodes"], cur_pdr * 100, "o", color="#dc2626", ms=10) # ёмкость при 95% PDR cap95 = nn[np.searchsorted(-pdr, -0.95)] if (pdr < 0.95).any() else nn[-1] ax_cap.axvline(cap95, color="#dc2626", ls=":", lw=1) ax_cap.set_xscale("log") ax_cap.set_xlabel("узлов на шлюз") ax_cap.set_ylabel("доля доставленных кадров PDR, %") ax_cap.set_ylim(0, 102) ax_cap.set_title(f"Ёмкость шлюза (ALOHA, {N_CHANNELS} каналов)\n" f"при {st['uplinks']:.1f} уплинк/сут: до {cap95:,.0f} узлов @95%".replace(",", " "), fontsize=9) ax_cap.grid(True, which="both", alpha=0.25) # --- ПРАВО: срок жизни узла от частоты уплинков --- ax_life.clear() up = np.logspace(-0.3, 3, 200) # 0.5 .. 1000 уплинков/сут life = np.array([node_life_years(u, sf)[0] for u in up]) ax_life.plot(up, life, color="#16a34a", lw=2) cur_life, q_day, q_sleep, q_active = node_life_years(st["uplinks"], sf) ax_life.plot(st["uplinks"], cur_life, "o", color="#dc2626", ms=10) ax_life.axhline(10, color="#64748b", ls="--", lw=1) ax_life.text(0.6, 11, "10 лет", color="#64748b", fontsize=8) ax_life.set_xscale("log") ax_life.set_yscale("log") ax_life.set_xlabel("уплинков в сутки на узел") ax_life.set_ylabel("срок жизни от пары AA, лет") ax_life.set_title(f"Энергобюджет узла (Stop2 {I_SLEEP_UA:.0f} мкА)\n" f"событийный ЖКХ (~2/сут) → десятилетия; раз/мин → месяцы", fontsize=9) ax_life.grid(True, which="both", alpha=0.25) # --- ТЕКСТ: сводка текущей точки --- ax_txt.clear(); ax_txt.axis("off") duty_cap = duty_max_uplinks_per_day(sf) share_sleep = 100 * q_sleep / q_day verdict = ("СОБЫТИЙНАЯ модель ЖКХ — сеть города реальна" if (cur_pdr >= 0.95 and cur_life >= 8) else "ПЕРЕГРУЗ: слишком часто шлём — сеть/батарея не тянут город") lines = [ f"SF{sf}/BW125: airtime кадра 10 Б = {ta*1000:.0f} мс | лимит duty 1% = {duty_cap:,.0f} уплинк/сут на узел".replace(",", " "), f"Текущая точка: {st['nodes']:,} узлов × {st['uplinks']:.1f} уплинк/сут → PDR {cur_pdr*100:.1f}% | G={G:.3f} Эрл/канал".replace(",", " "), f"Энергия: {q_day*1000:.1f} мкА·ч/сут (сон {share_sleep:.0f}%, эфир {100-share_sleep:.0f}%) → жизнь ≈ {cur_life:.1f} лет", f"ВЫВОД: {verdict}", ] colors = ["#334155", "#334155", "#334155", "#16a34a" if "реальна" in verdict else "#dc2626"] for i, (ln, c) in enumerate(zip(lines, colors)): ax_txt.text(0.0, 0.85 - i * 0.28, ln, transform=ax_txt.transAxes, fontsize=10, color=c, fontweight="bold" if i == 3 else "normal", family="monospace") fig.canvas.draw_idle() ax_n = fig.add_axes([0.15, 0.11, 0.70, 0.025]) s_n = Slider(ax_n, "узлов на шлюз", 100, 100000, valinit=st["nodes"], valstep=100) ax_u = fig.add_axes([0.15, 0.07, 0.70, 0.025]) s_u = Slider(ax_u, "уплинков в сутки на узел", 0.5, 1440, valinit=st["uplinks"], valstep=0.5) ax_s = fig.add_axes([0.15, 0.03, 0.70, 0.025]) s_s = Slider(ax_s, "SF (дальность vs airtime)", 7, 12, valinit=st["sf"], valstep=1) def on_n(v): st["nodes"] = int(v); redraw() def on_u(v): st["uplinks"] = v; redraw() def on_s(v): st["sf"] = int(v); redraw() s_n.on_changed(on_n); s_u.on_changed(on_u); s_s.on_changed(on_s) redraw() plt.show() if __name__ == "__main__": main()