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

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

Verbindet die Mitschnitte der Gehweg-Messung zu einer Raumtabelle:
  walk-ping.csv         (5 Pings/s auf die SONDE)     -> Verlust und laengste Luecke
  walk-ping-iphone.csv  (5 Pings/s auf die REFERENZ)  -> dito, seit 2026-08-22
  beide-telefone.csv    (AP + Signal beider Geraete)  -> seit 2026-08-22
  sampler.csv           (Funkdetail der Sonde)        -> Signal, PHY-Rate, Durchsatz
  markers.csv           (RAUM=-Klicks)                -> Raumgrenzen auf der Zeitachse

🔎 Zwei Geraete, zwei Rollen (Messaufbau seit 2026-08-22):
   REFERENZ = iPhone, roamt frei -> was ein normaler Client hier wirklich erlebt.
   SONDE    = S22, per access-list auf den GEGEN-AP gezwungen -> was die Rueckfallebene
              an derselben Stelle noch leisten koennte.
   Nebeneinander beantworten sie die Frage, die eine Messung allein nicht kann:
   ist es hier gut, und was passiert, wenn der gute AP wegfaellt?

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 lade_pings(pfad):
    """(sekunde, seq) je Antwort. Verluste stehen als Luecken in den Sequenznummern."""
    treffer = []
    if not os.path.exists(pfad):
        return treffer
    for r in csv.DictReader(open(pfad)):
        if not (r.get("seq") or "").isdigit():
            continue                       # RECONNECT-Zeilen
        if (r.get("ms") or "").strip():
            treffer.append((sec(r["ts_local"]), int(r["seq"])))
    return treffer


def lade_pings_alt(pfad):
    """Altes Format: BSD-ping vom Laptop."""
    treffer = []
    for line in open(pfad):
        m = re.match(r"(\d\d:\d\d:\d\d) 64 bytes.*icmp_seq=(\d+)", line)
        if m:
            treffer.append((sec(m.group(1)), int(m.group(2))))
    return treffer


def geraete(run_dir):
    """Liste von Geraeten: (rolle, ping-Liste, funk-Liste).

    funk-Liste = (sekunde, iface, signal, phy_rate_oder_None). Die PHY-Rate gibt es nur
    fuer die Sonde (sampler.csv, 1-s-Takt); fuer die Referenz liefert beide-telefone.csv
    nur AP und Signal im 2-s-Takt. Das ist bewusst so und wird unten beim Ruhezustand
    beruecksichtigt -- ohne Rate ist der Befund schwaecher und wird auch so benannt.
    """
    pfad = lambda n: os.path.join(run_dir, n)

    sonde_funk = []
    if os.path.exists(pfad("sampler.csv")):
        for r in csv.DictReader(open(pfad("sampler.csv"))):
            if r.get("sta_signal"):
                sonde_funk.append((sec(r["ts_local"]), r.get("sta_iface") or "",
                                   int(r["sta_signal"]),
                                   int(r["sta_tx_rate"]) if (r.get("sta_tx_rate") or "").isdigit() else None))

    ref_funk = []
    if os.path.exists(pfad("beide-telefone.csv")):
        for r in csv.DictReader(open(pfad("beide-telefone.csv"))):
            if (r.get("iphone_sig") or "").lstrip("-").isdigit():
                ref_funk.append((sec(r["ts_local"]), r.get("iphone_iface") or "",
                                 int(r["iphone_sig"]), None))

    sonde_ping = lade_pings(pfad("walk-ping.csv"))
    if not sonde_ping and os.path.exists(pfad("walk-ping.log")):
        sonde_ping = lade_pings_alt(pfad("walk-ping.log"))
    ref_ping = lade_pings(pfad("walk-ping-iphone.csv"))

    if not ref_ping and not ref_funk:
        # Lauf vor 2026-08-22: nur ein Geraet. Welches, ist aus den Dateien nicht
        # ableitbar -- also wird es auch nicht behauptet.
        return [("Gerät", sonde_ping, sonde_funk)]
    return [("Referenz (frei)", ref_ping, ref_funk),
            ("Sonde (Gegen-AP)", sonde_ping, sonde_funk)]


def auswerten(ping, funk, start, stop):
    """Verlust, laengste Luecke, Signal und APs eines Geraets in einem Zeitfenster."""
    p = [x for x in ping if start <= x[0] < stop]
    f = [x for x in funk if start <= x[0] < stop]
    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
    sig = [x[2] for x in f]
    raten = [x[3] for x in f if x[3]]
    # 🔴 Ruhezustand von Funkloch trennen: bleibt das Signal gut UND die PHY-Rate hoch,
    #    war das Geraet erreichbar und wollte nur nicht antworten (iOS bei gesperrtem
    #    Display). Ohne Ratendaten ist das nur ein Verdacht -- und wird so beschriftet.
    gut = bool(sig) and (sum(sig) / len(sig)) > -68
    schlaefrig = luecke >= 3 and gut and bool(raten) and (sum(raten) / len(raten)) > 100e6
    verdacht = luecke >= 3 and gut and not raten
    # 🔴 "Funk da, aber KEINE einzige Ping-Antwort" ist ein eigener Zustand und darf nicht
    #    als "—" (= keine Daten) erscheinen. Beim Abnahmetest am 2026-08-22 lag das iPhone
    #    gesperrt auf dem Tisch: angemeldet mit -50 dBm, und trotzdem stumm. Wer das als
    #    Datenluecke liest, sucht anschliessend ein Funkloch, das es nicht gibt.
    stumm = (len(p) <= 1) and bool(sig)
    return {
        "sig": sig, "luecke": luecke, "verloren": verloren, "gesamt": gesamt,
        "ifaces": sorted({AP.get(x[1], x[1]) for x in f if x[1]}),
        "schlaefrig": schlaefrig, "verdacht": verdacht, "stumm": stumm, "gut": gut,
        "leer": not p and not f,
    }


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

    global AP
    AP = ap_karte(a.dir)

    gs = geraete(a.dir)
    if not any(g[1] or g[2] for g in gs):
        raise SystemExit("Weder Ping-Antworten noch Funkproben gefunden.")

    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((sec(m["ts_local"]), mk[5:].title()))
    if not rooms:
        raise SystemExit("Keine RAUM=-Marken in markers.csv — nichts zuzuordnen.")

    ende = max([x[0] for g in gs for x in (g[1] + [(0, 0)])] +
               [x[0] for g in gs for x in (g[2] + [(0, 0, 0, None)])]) + 1
    fenster = [(r[0], rooms[i + 1][0] if i + 1 < len(rooms) else ende, r[1])
               for i, r in enumerate(rooms)]

    mehrere = len(gs) > 1
    if mehrere:
        print("REFERENZ = iPhone, roamt frei (was du hier wirklich hast)")
        print("SONDE    = S22, auf den Gegen-AP gezwungen (was die Rückfallebene noch kann)\n")
    print("%-15s %-17s %5s  %-22s %-17s %s"
          % ("Raum", "Rolle", "Dauer", "Signal min/⌀/max", "Verlust", "längste Lücke"))
    print("-" * 100)

    for start_s, stop_s, name in fenster:
        werte = [(rolle, auswerten(ping, funk, start_s, stop_s)) for rolle, ping, funk in gs]
        if all(w["leer"] for _, w in werte):
            continue
        erste = True
        for rolle, w in werte:
            if w["leer"]:
                continue
            sig, luecke = w["sig"], w["luecke"]
            if w["stumm"]:
                lz = "⚠️ Display gesperrt?" if w["gut"] else "⚠️ keine Antwort"
            elif luecke and w["schlaefrig"]:
                lz = "%.1f s ⚠️ eher Ruhezustand" % luecke
            elif luecke and w["verdacht"]:
                lz = "%.1f s ⚠️ evtl. Ruhezustand" % luecke
            elif luecke:
                lz = "%.1f s" % luecke
            else:
                lz = "keine"
            print("%-15s %-17s %4ds  %-22s %-17s %s" % (
                name if erste else "", rolle, stop_s - start_s,
                "%d / %.1f / %d dBm" % (min(sig), sum(sig) / len(sig), max(sig)) if sig else "—",
                ("%d von %d (%.0f %%)" % (w["verloren"], w["gesamt"],
                                          100 * w["verloren"] / w["gesamt"]) if w["gesamt"]
                 else ("keine Antwort" if w["stumm"] else "—")),
                lz))
            if w["ifaces"]:
                print("%-15s %-17s        auf %s" % ("", "", ", ".join(w["ifaces"])))
            erste = False

    print("\n⚠️ eher Ruhezustand = Lücke bei gutem Signal UND hoher PHY-Rate -> das Gerät war")
    print("   erreichbar und hat nur nicht geantwortet (iOS mit gesperrtem Display).")
    print("⚠️ evtl. Ruhezustand = dasselbe Muster, aber ohne PHY-Rate zum Gegenprüfen")
    print("   (die Referenz liefert nur AP und Signal). Verdacht, kein Befund.")
    print("⚠️ Display gesperrt? = Funkverbindung stand mit gutem Signal, aber KEINE einzige")
    print("   Ping-Antwort. Das ist kein Funkloch und keine Datenlücke — bei iOS beantwortet")
    print("   ein gesperrtes Display kein ICMP. Beim nächsten Lauf Display anlassen.")
    print("\nLesehilfe: eine längste Lücke über ~2 s ist für einen Videocall eine hörbare")
    print("Unterbrechung, über 10 s ist der Call praktisch tot. Signal unter −70 dBm ist die")
    print("Zone, in der das passiert.")
    if mehrere:
        print("\nDie interessante Spalte ist der Unterschied: gute Referenz + schlechte Sonde")
        print("= hier hängt alles an EINEM AP. Beide schlecht = echtes Loch ohne Rückfallebene.")


if __name__ == "__main__":
    main()