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 sys
import urllib.parse


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


# 🔴🔴 Hier stand bis 2026-08-22 eine feste Tabelle -- mit VERTAUSCHTEN APs: cap-wifi1/2
#     sind ap2, nicht ap1 (live belegt ueber cap=, configuration= und Kanal). Die Baender
#     stimmten, die Identitaet nicht. Der Manager vergibt die cap-wifiN-Nummern DYNAMISCH in
#     der Reihenfolge, in der sich die CAPs anmelden -- eine Konstante dazu ist nur bis zum
#     naechsten AP-Reboot richtig und faellt danach still falsch aus.
#
#     ⚠️ OFFEN: ob die Messungen vom 2026-08-21 (Raumkarte, Sideboard-Test) mit vertauschten
#     AP-Namen ausgewertet wurden, ist NICHT geprueft. Die Rohdaten sind unberuehrt -- es geht
#     allein um die Beschriftung. Betroffen waere u.a. Fund F ("Schlafzimmer -61 ap1 -> -83
#     ap2") und damit die Sideboard-Entscheidung. Vor dem Zitieren alter Zahlen klaeren.
def ap_karte(run_dir):
    """cap-wifiN -> apX/Band, aus der Karte DIESES Messlaufs (ap-map.txt).

    🔴 Ausdruecklich NICHT live nachschlagen. Die Nummern gelten nur fuer den Zeitpunkt der
       Messung; ein spaeter erzeugtes Mapping auf alte Daten anzuwenden ergibt eine
       selbstbewusst falsche Antwort. Fehlt die Datei (alle Laeufe vor 2026-08-22), bleiben
       die rohen cap-wifiN stehen -- unbeschriftet ist ehrlicher als falsch beschriftet.
    """
    pfad = os.path.join(run_dir, "ap-map.txt")
    if os.path.exists(pfad):
        with open(pfad) as fh:
            return dict(z.strip().split("=", 1) for z in fh if "=" in z)
    print("⚠️  Kein ap-map.txt in diesem Lauf -- AP-Namen bleiben roh (cap-wifiN).\n"
          "    Fuer Laeufe vor 2026-08-22 ist die Zuordnung NICHT rekonstruierbar: sie haengt\n"
          "    daran, in welcher Reihenfolge sich die CAPs damals angemeldet haben.\n"
          "    Anhaltspunkt aus der Physik: der AP im Raum sollte dort das staerkste Signal\n"
          "    liefern (ap2 steht in Technik OG, ap1 haengt an core ether15).",
          file=sys.stderr)
    return {}


AP = {}


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()

    global AP
    AP = ap_karte(a.dir)

    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()