bf-infra/docs/tp25-gehweg-auswertung.py

#!/usr/bin/env python3
"""
TP25 — Gehweg-Auswertung: Signal und Erreichbarkeit pro RAUM.

Verbindet die drei Mitschnitte der Gehweg-Messung zu einer Raumtabelle:
  walk-ping.log  (5 Pings/s auf das Telefon)  -> Verlust und laengste Luecke
  sampler.csv    (Funkwerte am Manager)       -> AP/Band und Signal
  markers.csv    (RAUM=-Klicks)               -> die Raumgrenzen auf der Zeitachse

Damit wird aus "irgendwo zwischen 15:07 und 15:09 war es schlecht" ein Satz mit Raumnamen.

    python3 docs/tp25-gehweg-auswertung.py --dir <run-dir>
"""
import argparse
import csv
import os
import re
import urllib.parse


def sec(t):
    h, m, s = map(int, t.split(":"))
    return h * 3600 + m * 60 + s


AP = {"cap-wifi1": "ap1/5GHz", "cap-wifi2": "ap1/2,4GHz",
      "cap-wifi3": "ap2/5GHz", "cap-wifi4": "ap2/2,4GHz"}


def ros_ms(v):
    """RouterOS-Dauer in Millisekunden. Schreibweisen: "12ms", "3ms441us", "529us", "1s200ms"."""
    total = 0.0
    for zahl, einheit in re.findall(r"(\d+)(us|ms|s)", v):
        total += int(zahl) * {"us": 0.001, "ms": 1.0, "s": 1000.0}[einheit]
    return total


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--dir", required=True)
    a = ap.parse_args()

    pings = []
    csv_path = os.path.join(a.dir, "walk-ping.csv")
    if os.path.exists(csv_path):
        # Neues Format: Ping vom Core, "ts_local,seq,ms" — leeres ms = Verlust.
        for r in csv.DictReader(open(csv_path)):
            if not (r.get("seq") or "").isdigit():
                continue           # RECONNECT-Zeilen
            ms = (r.get("ms") or "").strip()
            if ms:
                pings.append((sec(r["ts_local"]), int(r["seq"]), ros_ms(ms)))
    else:
        # Altes Format: BSD-ping vom Laptop.
        for line in open(os.path.join(a.dir, "walk-ping.log")):
            m = re.match(r"(\d\d:\d\d:\d\d) 64 bytes.*icmp_seq=(\d+).*time=([\d.]+)", line)
            if m:
                pings.append((sec(m.group(1)), int(m.group(2)), float(m.group(3))))
    if not pings:
        raise SystemExit("Keine Ping-Antworten gefunden.")

    rows = [r for r in csv.DictReader(open(os.path.join(a.dir, "sampler.csv")))
            if r.get("sta_signal")]

    rooms = []
    for m in csv.DictReader(open(os.path.join(a.dir, "markers.csv"))):
        # 🔴 Marker kommen URL-kodiert aus dem Formular ("RAUM%3DDIELE%20UG") — dekodieren,
        #    sonst heisst der Raum "%3DDIELE%20UG" und die Tabelle ist unlesbar.
        mk = urllib.parse.unquote(m["marker"])
        if mk.upper().startswith("RAUM="):
            rooms.append((int(m["ts_epoch"]), sec(m["ts_local"]), mk[5:].title()))
    if not rooms:
        raise SystemExit("Keine RAUM=-Marken in markers.csv — nichts zuzuordnen.")

    ende = max(pings[-1][0], sec(rows[-1]["ts_local"]) if rows else 0) + 1
    fenster = [(r[1], rooms[i + 1][1] if i + 1 < len(rooms) else ende, r[2])
               for i, r in enumerate(rooms)]

    print("%-16s %6s  %-24s %-18s %s" % ("Raum", "Dauer", "Signal min/avg/max", "Verlust", "längste Lücke"))
    print("-" * 92)
    for start, stop, name in fenster:
        p = [x for x in pings if start <= x[0] < stop]
        r = [x for x in rows if start <= sec(x["ts_local"]) < stop]
        if not p and not r:
            continue
        # Verlust und laengste Luecke aus den Sequenznummern
        luecke, verloren = 0.0, 0
        for i in range(1, len(p)):
            miss = p[i][1] - p[i - 1][1] - 1
            if miss > 0:
                verloren += miss
                luecke = max(luecke, miss * 0.2)
        gesamt = (p[-1][1] - p[0][1] + 1) if len(p) > 1 else 0
        # 🔴 Ruhezustand von Funkloch trennen. Ein iPhone mit gesperrtem Display
        #    beantwortet ICMP nicht — das sieht in der Verlustspalte aus wie ein Funkloch.
        #    Unterscheidbar ist es am Funk: bleibt das Signal gut UND die PHY-Rate hoch,
        #    war das Geraet erreichbar und wollte nur nicht antworten.
        #    (Am 2026-08-21 auf der Terrasse: 58 s Ping-Luecke bei -53 dBm und 275 Mbps.)
        sig = [int(x["sta_signal"]) for x in r]
        raten = [int(x["sta_tx_rate"] or 0) for x in r if x.get("sta_tx_rate")]
        #    Bewertet wird der DURCHSCHNITT des Abschnitts, nicht sein Minimum: ein
        #    einzelner schwacher Sample beim Durchgehen darf den Befund nicht kippen.
        schlaefrig = (luecke >= 3 and sig and (sum(sig) / len(sig)) > -68
                      and raten and (sum(raten) / len(raten)) > 100e6)
        ifaces = sorted({AP.get(x["sta_iface"], x["sta_iface"]) for x in r if x["sta_iface"]})
        print("%-16s %5ds  %-24s %-18s %s" % (
            name, stop - start,
            "%d / %.1f / %d dBm" % (min(sig), sum(sig) / len(sig), max(sig)) if sig else "—",
            "%d von %d (%.0f %%)" % (verloren, gesamt, 100 * verloren / gesamt) if gesamt else "—",
            (("%.1f s ⚠️ eher Ruhezustand" % luecke) if schlaefrig
             else ("%.1f s" % luecke)) if luecke else "keine"))
        if ifaces:
            print("%-16s        auf %s" % ("", ", ".join(ifaces)))
    print("\n⚠️ = Luecke bei gutem Signal und hoher PHY-Rate -> das Geraet war erreichbar und hat")
    print("   nur nicht geantwortet (iOS-Ruhezustand). Solche Luecken NICHT als Funkloch lesen.")
    print("\nLesehilfe: eine längste Lücke über ~2 s ist für einen Videocall eine hörbare Unterbrechung,")
    print("über 10 s ist der Call praktisch tot. Signal unter −70 dBm ist die Zone, in der das passiert.")


if __name__ == "__main__":
    main()