Christoph Emsenhuber
66035c3166
Pingt der messende Laptop das Telefon an, laeuft der Weg ueber zwei Funkstrecken -- Laptop -> AP -> Core -> AP -> Telefon. Fehlt dann ein Paket, ist nicht entscheidbar, welches der beiden Geraete weg war, und genau diese Trennschaerfe ist der ganze Zweck der Messung. Vom Core aus ist die Strecke bis zum AP durchgehend Kabel. Ausloeser war unfreiwillig: das MacBook ist mitten in einem Lauf vom Dock ans WLAN gewechselt und hat dabei die IP gewechselt -- die Konsole war ploetzlich nicht mehr erreichbar, und der Ping haette ab da zwei Funkstrecken gemessen. Das Ping-Protokoll ist jetzt eine CSV (ts_local,seq,ms; leeres ms = Verlust), die Auswertung liest beide Formate, damit die Laeufe von gestern lesbar bleiben. RouterOS-Dauern kommen als "3ms441us" oder "529us" und werden entsprechend geparst. Eine Falle steckt im Skript-Kommentar: tr puffert blockweise, sobald die Ausgabe in eine Pipe geht -- die ersten ~130 Zeilen haengen im Puffer und die Datei bleibt scheinbar leer. Das \r kommt deshalb im Shell-Loop weg, nicht per tr. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
105 lines
4.1 KiB
Python
105 lines
4.1 KiB
Python
#!/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
|
||
|
||
|
||
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"))):
|
||
if m["marker"].startswith("RAUM="):
|
||
rooms.append((int(m["ts_epoch"]), sec(m["ts_local"]), m["marker"][5:]))
|
||
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
|
||
sig = [int(x["sta_signal"]) for x in r]
|
||
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" % luecke) if luecke else "keine"))
|
||
if ifaces:
|
||
print("%-16s auf %s" % ("", ", ".join(ifaces)))
|
||
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()
|