Sound server added

2026-06-16 17:36:21 +02:00
parent 528c7df10e
commit fdbbc98088
8 changed files with 522 additions and 0 deletions
--- a/soundmeter_log.py
+++ b/soundmeter_log.py
@@ -0,0 +1,147 @@
+#!/usr/bin/env python3
+"""
+soundmeter_log.py — BLE sound-meter logger + iMX8 tone-server client.
+Target: Windows 11 / Python 3.10+.
+
+  pip install bleak requests
+
+For each test step:
+  1. POST /play to tone_server.py on the iMX8
+  2. Collect dB readings from the RS-95-EM BLE meter for the step duration
+  3. Write avg / min / max to CSV
+
+BLE packet format
+-----------------
+  Header : bytes[0]=0xD5  bytes[1]=0xF0
+  dB     : bytes[4:6]  uint16 big-endian / 10.0
+"""
+
+import asyncio
+import csv
+import datetime
+import struct
+from pathlib import Path
+
+import requests
+from bleak import BleakClient
+
+# ── Configuration ─────────────────────────────────────────────────────────────
+IMX8_IP = "10.32.34.103"
+IMX8_URL = f"http://{IMX8_IP}:5000"
+
+BLE_MAC  = "B0:D2:78:5C:20:07"
+BLE_CHAR = "0000fff2-0000-1000-8000-00805f9b34fb"
+
+# ── Test sequence — edit as required ─────────────────────────────────────────
+_FREQS = [100, 250, 500, 1000, 2000, 3000, 4000, 5000, 6000, 7000,
+          8000, 9000, 10000, 11000, 12000, 13000, 14000, 15000]
+
+TEST_STEPS = [
+    {"freq": freq, "vol": round(vol / 10, 1), "duration": 5.0}
+    for freq in _FREQS
+    for vol in range(0, 11)   # 0% → 100% in 10% steps
+]
+
+# ── Output ────────────────────────────────────────────────────────────────────
+_ts      = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
+CSV_FILE = Path(f"audio_test_{_ts}.csv")
+CSV_HDR  = ['timestamp', 'freq_hz', 'vol', 'duration_s',
+            'avg_db', 'min_db', 'max_db', 'samples']
+
+
+# ── BLE packet parser ─────────────────────────────────────────────────────────
+def _parse_db(data: bytearray) -> float | None:
+    if len(data) < 6 or data[0] != 0xD5 or data[1] != 0xF0:
+        return None
+    return struct.unpack_from('>H', data, 4)[0] / 10.0
+
+
+# ── HTTP helpers (blocking — run via asyncio.to_thread) ───────────────────────
+def _http_play(freq: float, vol: float, duration: float) -> dict:
+    r = requests.post(
+        f"{IMX8_URL}/play",
+        json={"freq": freq, "vol": vol, "duration": duration},
+        timeout=5,
+    )
+    r.raise_for_status()
+    return r.json()
+
+
+def _http_stop() -> dict:
+    r = requests.post(f"{IMX8_URL}/stop", timeout=5)
+    r.raise_for_status()
+    return r.json()
+
+
+# ── Main test runner ──────────────────────────────────────────────────────────
+async def run_test():
+    readings: list[float] = []
+
+    def on_notify(_sender, data: bytearray):
+        db = _parse_db(data)
+        if db is not None:
+            readings.append(db)
+
+    print(f"Connecting to BLE {BLE_MAC} …")
+    async with BleakClient(BLE_MAC, timeout=15.0) as client:
+        print("Connected. Enumerating services…")
+        for svc in client.services:
+            print(f"  SVC {svc.uuid}")
+            for ch in svc.characteristics:
+                props = ','.join(ch.properties)
+                print(f"    CHAR {ch.uuid}  [{props}]")
+
+        await client.start_notify(BLE_CHAR, on_notify)
+        print(f"\nSubscribed to {BLE_CHAR}\n")
+
+        with CSV_FILE.open('w', newline='') as f:
+            writer = csv.writer(f)
+            writer.writerow(CSV_HDR)
+
+            for i, step in enumerate(TEST_STEPS, 1):
+                freq = step['freq']
+                vol  = step['vol']
+                dur  = step['duration']
+                print(f"[{i}/{len(TEST_STEPS)}]  {freq:>5} Hz  vol={vol:.2f}  dur={dur:.1f}s  …", end='', flush=True)
+
+                try:
+                    await asyncio.to_thread(_http_play, freq, vol, dur)
+                except Exception as e:
+                    print(f"  ERROR /play: {e}")
+                    continue
+
+                readings.clear()
+                await asyncio.sleep(dur)
+
+                snap = list(readings)
+                if snap:
+                    avg_db = sum(snap) / len(snap)
+                    min_db = min(snap)
+                    max_db = max(snap)
+                    print(f"  avg={avg_db:5.1f} dB  min={min_db:5.1f}  max={max_db:5.1f}  n={len(snap)}")
+                else:
+                    avg_db = min_db = max_db = 0.0
+                    print("  WARNING: no BLE readings received")
+
+                ts = datetime.datetime.now().isoformat(timespec='seconds')
+                writer.writerow([
+                    ts, freq, vol, dur,
+                    f"{avg_db:.1f}", f"{min_db:.1f}", f"{max_db:.1f}", len(snap),
+                ])
+                f.flush()
+
+                await asyncio.sleep(0.5)   # brief gap between steps
+
+        # Ensure tone is off
+        try:
+            await asyncio.to_thread(_http_stop)
+        except Exception:
+            pass
+
+        await client.stop_notify(BLE_CHAR)
+
+    print(f"\nResults saved → {CSV_FILE.resolve()}")
+
+
+if __name__ == '__main__':
+    asyncio.run(run_test())