#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ warwalk_csv2map.py — путь А (офлайн): лог консоли ходока -> карта и/или SQL. Вход: файл, куда телефон/ноутбук писал консоль ходока (115200). Из него берутся строки: GEO,millis,valid,lat_e7,lon_e7,hdop_x10,sats,heard,rssi_med,snr,win_ok,win_bad MARK,millis Остальные строки (STAT, [GPS], баннеры) молча игнорируются — лог можно скармливать как есть, не вычищая. Выходы: --html map.html интерактивная карта (folium): точки, окрашенные по RSSI, чёрные точки-дыры (heard=0), флажки MARK, опц. слой heatmap --sql out.sql готовые INSERT'ы в таблицу warwalk (та же схема, что у живого пути Б) -> psql -U postgres -d iot_sensors -f out.sql и офлайн-прогулка появляется в том же Grafana-дашборде. Примеры: python3 warwalk_csv2map.py walk.log --html walk.html python3 warwalk_csv2map.py walk.log --html walk.html --heatmap python3 warwalk_csv2map.py walk.log --sql walk.sql --session 2026-07-12-2dbm Честное предупреждение про --heatmap: folium.HeatMap — это КАРТА ПЛОТНОСТИ со взвешиванием, а не интерполяция поля. Где топтались дольше — там «горячее», даже при слабом сигнале. Поэтому скрипт перед heatmap прореживает точки в сетку ~10 м и берёт медиану по ячейке. Для честной интерполяции (IDW/кригинг) экспортируй --sql/CSV в QGIS — см. райтап, раздел «Визуализация». """ import argparse import sys from datetime import datetime, timedelta RSSI_MIN, RSSI_MAX = -125.0, -50.0 # шкала окраски, дБм GRID = 4 # округление градусов для сетки ~10 м def parse_log(path, min_hdop, require_valid): """-> (points, marks_millis). point = dict.""" points, marks = [], [] with open(path, "r", encoding="utf-8", errors="replace") as f: for lineno, raw in enumerate(f, 1): line = raw.strip() if line.startswith("MARK,"): try: marks.append(int(line.split(",")[1])) except (IndexError, ValueError): pass continue if not line.startswith("GEO,"): continue parts = line.split(",") if len(parts) != 12: print(f" ! строка {lineno}: {len(parts)} полей вместо 12, пропускаю", file=sys.stderr) continue try: p = { "millis": int(parts[1]), "valid": parts[2] == "1", "lat": int(parts[3]) / 1e7, "lon": int(parts[4]) / 1e7, "hdop": int(parts[5]) / 10.0, "sats": int(parts[6]), "heard": parts[7] == "1", "rssi": float(parts[8]), "snr": float(parts[9]), "win_ok": int(parts[10]), "win_bad": int(parts[11]), "mark": False, } except ValueError: print(f" ! строка {lineno}: мусор в числах, пропускаю", file=sys.stderr) continue if require_valid and not p["valid"]: continue if p["hdop"] > min_hdop: continue if p["lat"] == 0 and p["lon"] == 0: continue points.append(p) # MARK -> ближайшая по millis точка for m in marks: if points: min(points, key=lambda q: abs(q["millis"] - m))["mark"] = True return points, marks def rssi_color(rssi): """красный (-125) -> жёлтый -> зелёный (-50).""" t = max(0.0, min(1.0, (rssi - RSSI_MIN) / (RSSI_MAX - RSSI_MIN))) if t < 0.5: r, g = 255, int(510 * t) else: r, g = int(510 * (1 - t)), 255 return f"#{r:02x}{g:02x}00" def grid_median(points, key="rssi"): """сетка ~10 м, медиана по ячейке: лечим «где топтались — там горячее».""" cells = {} for p in points: cells.setdefault((round(p["lat"], GRID), round(p["lon"], GRID)), []).append(p[key]) out = [] for (lat, lon), vals in cells.items(): vals.sort() out.append((lat, lon, vals[len(vals) // 2])) return out def write_html(points, path, want_heatmap): try: import folium except ImportError: sys.exit("Нужен folium: pip install folium (тайлы OSM требуют интернета при просмотре)") heard = [p for p in points if p["heard"]] holes = [p for p in points if not p["heard"]] center = heard[0] if heard else points[0] fmap = folium.Map(location=[center["lat"], center["lon"]], zoom_start=16, control_scale=True) lg_pts = folium.FeatureGroup(name="RSSI маячка (медиана)").add_to(fmap) for p in heard: folium.CircleMarker( [p["lat"], p["lon"]], radius=5, color=rssi_color(p["rssi"]), fill=True, fill_opacity=0.85, weight=1, popup=f"{p['rssi']:.0f} дБм, SNR {p['snr']:.1f}, n={p['win_ok']}, HDOP {p['hdop']:.1f}", ).add_to(lg_pts) lg_holes = folium.FeatureGroup(name="ДЫРЫ (пакетов нет)").add_to(fmap) for p in holes: folium.CircleMarker([p["lat"], p["lon"]], radius=4, color="#000000", fill=True, fill_opacity=0.9, weight=1, popup="дыра покрытия (0 пакетов в окне)").add_to(lg_holes) lg_marks = folium.FeatureGroup(name="MARK (кнопка)").add_to(fmap) for p in points: if p["mark"]: folium.Marker([p["lat"], p["lon"]], icon=folium.Icon(color="blue", icon="flag"), popup=f"MARK @ {p['millis']} мс").add_to(lg_marks) if want_heatmap and heard: from folium.plugins import HeatMap cells = grid_median(heard) data = [[lat, lon, max(0.0, min(1.0, (v - RSSI_MIN) / (RSSI_MAX - RSSI_MIN)))] for lat, lon, v in cells] HeatMap(data, name="heatmap (сетка 10 м, медиана)", radius=18, blur=22, min_opacity=0.3).add_to(fmap) folium.LayerControl(collapsed=False).add_to(fmap) fmap.save(path) print(f"Карта: {path} (точек: {len(heard)}, дыр: {len(holes)})") def write_sql(points, path, session, walker_id, t0): """millis -> абсолютное время от t0 (старт лога); INSERT'ы в warwalk.""" base = points[0]["millis"] if points else 0 with open(path, "w", encoding="utf-8") as f: f.write("-- офлайн-импорт WarWalk (путь А); psql -U postgres -d iot_sensors -f этот_файл\n") f.write("BEGIN;\n") for p in points: t = t0 + timedelta(milliseconds=p["millis"] - base) rssi = str(p["rssi"]) if p["heard"] else "NULL" snr = str(p["snr"]) if p["heard"] else "NULL" f.write( "INSERT INTO warwalk (time, walker_id, session, lat, lon, hdop, sats, gps_valid," " rssi_fox, snr_fox, n_fox, mark) VALUES " f"('{t.isoformat()}', {walker_id}, '{session}', {p['lat']:.7f}, {p['lon']:.7f}," f" {p['hdop']:.1f}, {p['sats']}, {str(p['valid']).lower()}," f" {rssi}, {snr}, {p['win_ok']}, {str(p['mark']).lower()});\n" ) f.write("COMMIT;\n") print(f"SQL: {path} (строк: {len(points)}, session: {session})") def main(): ap = argparse.ArgumentParser(description="Лог ходока WarWalk -> карта / SQL") ap.add_argument("log", help="файл с консольным логом ходока") ap.add_argument("--html", help="выход: интерактивная карта HTML (folium)") ap.add_argument("--heatmap", action="store_true", help="добавить heatmap-слой к --html") ap.add_argument("--sql", help="выход: INSERT'ы в таблицу warwalk") ap.add_argument("--session", default=datetime.now().strftime("%Y-%m-%d-offline"), help="метка прогулки для --sql") ap.add_argument("--walker-id", type=int, default=2) ap.add_argument("--start", help="реальное время старта лога, ISO (иначе — сейчас минус длительность)") ap.add_argument("--max-hdop", type=float, default=3.5, help="фильтр качества GPS") ap.add_argument("--keep-invalid", action="store_true", help="не выбрасывать точки с valid=0 (по умолчанию выбрасываются)") args = ap.parse_args() points, marks = parse_log(args.log, args.max_hdop, not args.keep_invalid) if not points: sys.exit("Ни одной пригодной GEO-точки: проверь valid/HDOP (--keep-invalid?) и сам лог.") print(f"Точек: {len(points)}, из них дыр: {sum(1 for p in points if not p['heard'])}, MARK: {len(marks)}") if args.html: write_html(points, args.html, args.heatmap) if args.sql: if args.start: t0 = datetime.fromisoformat(args.start) else: dur = points[-1]["millis"] - points[0]["millis"] t0 = datetime.now().astimezone() - timedelta(milliseconds=dur) write_sql(points, args.sql, args.session, args.walker_id, t0) if not args.html and not args.sql: print("Ничего не сделано: укажи --html и/или --sql.") if __name__ == "__main__": main()