TP25/TP28: Walk-Harness auf N Geräte, Analyzer generalisiert, AP-Map-Fix

- tp25-walk-messung.sh: halber Refactor geschlossen — Rumpf schleift ueber die
  PH_*-Arrays (pro Geraet ein Sampler + ein Core-Ping, sampler-<name>.csv /
  walk-ping-<name>.csv), PID-Arrays im Cleanup, Archiv-/Echo-Block generisch.
  Lief vorher unter set -u gar nicht (unbound $PHONE_MAC).
- tp25-gehweg-auswertung.py: liest N benannte Geraete per glob; altes 2-Rollen-
  Schema (sampler.csv/walk-ping.csv/walk-ping-iphone.csv/beide-telefone.csv) bleibt
  als Legacy-Fallback erhalten. Beide Pfade getestet.
- tp25-pin-s22.sh: karte() repariert — channel.frequency ist seit dem DFS-Kanalplan
  (2026-09-03, Frequenz-Listen) ein Array; der alte Skalar-Vergleich blankte map,
  wodurch ap-map.txt leer blieb. Fix nimmt bei :typeof=array das erste Element.

Live-Smoke (4 Geraete) gruen: alle erreichbar, sauberes Cleanup, korrekte AP-Map.
Naechster Schritt: In-situ-Walk (nur Zahlen, Commit/Rollback des Kanalplans separat).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-09-04 12:42:30 +02:00
parent 45e4540cf5
commit ac3a125c82
3 changed files with 105 additions and 39 deletions

View File

@@ -22,6 +22,7 @@ Damit wird aus "irgendwo zwischen 15:07 und 15:09 war es schlecht" ein Satz mit
"""
import argparse
import csv
import glob
import os
import re
import sys
@@ -98,24 +99,45 @@ def lade_pings_alt(pfad):
return treffer
def geraete(run_dir):
"""Liste von Geraeten: (rolle, ping-Liste, funk-Liste).
def _sampler_funk(pfad):
"""funk-Liste (sekunde, iface, signal, phy_rate_oder_None) aus einer sampler-CSV."""
funk = []
if os.path.exists(pfad):
for r in csv.DictReader(open(pfad)):
if r.get("sta_signal"):
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))
return funk
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.
def geraete(run_dir):
"""Liste benannter Geraete: (name, ping-Liste, funk-Liste).
funk-Liste = (sekunde, iface, signal, phy_rate_oder_None).
Seit 2026-09-04 (N-Geraete-Umbau) schreibt die Harness pro Geraet walk-ping-<name>.csv
UND sampler-<name>.csv -- jedes Paar wird als EIN benanntes Geraet gelesen, beliebig viele,
ohne feste Sonde/Referenz-Rollen. Faellt auf das alte 2-Rollen-Schema zurueck (sampler.csv /
walk-ping.csv / walk-ping-iphone.csv / beide-telefone.csv), damit Laeufe vor dem Umbau
weiter auswertbar bleiben. Unterschieden wird an der NACKTEN sampler.csv/walk-ping.csv:
die schreibt nur das alte Schema -- die neue Harness nie (immer mit -<name>). Ohne diese
Klammer wuerde der glob unten die walk-ping-iphone.csv eines Alt-Laufs fassen und dessen
Sonde (walk-ping.csv, ohne Suffix) stillschweigend verlieren.
"""
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))
legacy = os.path.exists(pfad("sampler.csv")) or os.path.exists(pfad("walk-ping.csv"))
if not legacy:
gs = []
for pp in sorted(glob.glob(os.path.join(run_dir, "walk-ping-*.csv"))):
name = os.path.basename(pp)[len("walk-ping-"):-len(".csv")]
gs.append((name, lade_pings(pp), _sampler_funk(pfad("sampler-%s.csv" % name))))
if gs:
return gs
# --- Legacy: 2-Rollen-Schema (Laeufe vor 2026-09-04) ---
sonde_funk = _sampler_funk(pfad("sampler.csv"))
ref_funk = []
if os.path.exists(pfad("beide-telefone.csv")):
for r in csv.DictReader(open(pfad("beide-telefone.csv"))):
@@ -189,6 +211,9 @@ def main():
gs = geraete(a.dir)
if not any(g[1] or g[2] for g in gs):
raise SystemExit("Weder Ping-Antworten noch Funkproben gefunden.")
# Alte 2-Rollen-Laeufe tragen die Namen "Referenz"/"Sonde" und bekommen den Rollen-Text;
# neue N-Geraete-Laeufe tragen Geraetenamen (iphone/s22/ipad/...) -- ohne Rollen-Erklaerung.
legacy_rollen = any("Sonde" in r or "Referenz" in r for r, _, _ in gs)
rooms = []
for m in csv.DictReader(open(os.path.join(a.dir, "markers.csv"))):
@@ -209,9 +234,12 @@ def main():
for i, r in enumerate(rooms)]
mehrere = len(gs) > 1
if mehrere:
if mehrere and legacy_rollen:
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")
elif mehrere:
print("Ein Block je Raum, eine Zeile je Gerät (%s) — alle frei roamend.\n"
% ", ".join(r for r, _, _ in gs))
print("%-15s %-17s %5s %-22s %-17s %s"
% ("Raum", "Rolle", "Dauer", "Signal min/⌀/max", "Verlust", "längste Lücke"))
print("-" * 100)
@@ -262,9 +290,13 @@ def main():
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:
if mehrere and legacy_rollen:
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.")
elif mehrere:
print("\nDie interessante Spalte ist der Unterschied ZWISCHEN den Geräten am selben Ort:")
print("streuen Signal/Verlust stark, liegt es am Gerät (Antenne/Standby), nicht am Funk;")
print("sind alle an einem Ort schlecht, ist es ein echtes Loch der Ausleuchtung.")
if __name__ == "__main__":