publish eve_audio_watcher.py

eve_audio_watcher.py

@@ -0,0 +1,186 @@
+#!/usr/bin/env python3
+"""Audio watcher — catches things the game only tells you by SOUND.
+
+A bare target-lock is NOT in any log or API, but EVE plays a distinct sound when you
+get locked. This captures your Windows speaker output (WASAPI loopback) and matches it
+against short sound 'fingerprints' you record once — so it can warn you "someone locked
+you", "rock depleted", "warp/jump/dock", etc., that nothing else can see.
+
+How it learns (one-time per sound, do it while that sound plays in-game):
+    python eve_audio_watcher.py --learn targeted    # records ~2.5s, saves a fingerprint
+    python eve_audio_watcher.py --learn depleted
+    python eve_audio_watcher.py --list              # show learned sounds
+Then it runs headless during a mining session and alerts when it hears them.
+
+Deps (auto-installed by update.py): numpy, soundcard. If they're missing it exits
+quietly so it never breaks the rest of the watcher fleet.
+
+  python eve_audio_watcher.py            # run (waits for !mining on)
+"""
+import json
+import os
+import sys
+import time
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+import eve_orehold_watcher as w        # notify, bot_mining, bot_muted, heartbeat
+
+SOUNDS_DIR = os.path.join(HERE, "sounds")
+RATE = 48000
+LEARN_SECS = 2.5
+WINDOW_SECS = 1.2
+BANDS = 32
+MATCH_THRESHOLD = 0.86
+RMS_FLOOR = 0.004                       # ignore near-silence
+
+# preset alert text for the well-known cues (used when you --learn one of these names)
+PRESETS = {
+    "targeted": ("🔒 Targeted", "Someone just locked you — watch for tackle, be ready to warp.", "urgent"),
+    "depleted": ("Rock depleted", "Asteroid mined out — switch rocks.", "high"),
+    "warp": ("Warp", "Warp drive active.", "low"),
+    "jump": ("Jumped", "Gate jump.", "low"),
+    "dock": ("Docked", "Docked up.", "low"),
+    "undock": ("Undocked", "In space.", "low"),
+}
+
+
+def _np():
+    import numpy as np
+    return np
+
+
+def fingerprint(mono, rate=RATE, bands=BANDS):
+    """Spectral signature: log-spaced FFT band energies, L2-normalized. Length-agnostic."""
+    np = _np()
+    x = np.asarray(mono, dtype=np.float64).ravel()
+    if x.size < 256:
+        return None
+    x = x - x.mean()
+    X = np.abs(np.fft.rfft(x * np.hanning(x.size)))
+    freqs = np.fft.rfftfreq(x.size, 1.0 / rate)
+    edges = np.logspace(np.log10(80), np.log10(rate / 2), bands + 1)
+    feat = np.array([X[(freqs >= edges[i]) & (freqs < edges[i + 1])].mean()
+                     if np.any((freqs >= edges[i]) & (freqs < edges[i + 1])) else 0.0
+                     for i in range(bands)])
+    k = np.array([0.5, 1.0, 0.5])              # light blur -> robust to small pitch shift
+    feat = np.convolve(feat, k / k.sum(), mode="same")
+    n = np.linalg.norm(feat)
+    return feat / n if n > 0 else feat
+
+
+def rms(mono):
+    np = _np()
+    x = np.asarray(mono, dtype=np.float64).ravel()
+    return float(np.sqrt(np.mean(x * x))) if x.size else 0.0
+
+
+def cosine(a, b):
+    return float(_np().dot(a, b))      # inputs are unit vectors
+
+
+def load_templates():
+    out = {}
+    if not os.path.isdir(SOUNDS_DIR):
+        return out
+    np = _np()
+    for f in os.listdir(SOUNDS_DIR):
+        if f.endswith(".json"):
+            d = json.load(open(os.path.join(SOUNDS_DIR, f)))
+            out[d["name"]] = {**d, "vec": np.array(d["vec"])}
+    return out
+
+
+def _recorder(secs):
+    """Yield a mono numpy array of `secs` of speaker loopback."""
+    import soundcard as sc
+    np = _np()
+    spk = sc.default_speaker()
+    mic = sc.get_microphone(str(spk.name), include_loopback=True)
+    with mic.recorder(samplerate=RATE, channels=1) as rec:
+        data = rec.record(numframes=int(secs * RATE))
+    return np.asarray(data).ravel()
+
+
+def learn(name):
+    os.makedirs(SOUNDS_DIR, exist_ok=True)
+    print(f"Recording '{name}' for {LEARN_SECS}s — make the sound play NOW...")
+    vec = fingerprint(_recorder(LEARN_SECS))
+    if vec is None:
+        print("Got no audio. Is something playing? Try again.")
+        return
+    title, msg, prio = PRESETS.get(name, (name.title(), f"Heard: {name}", "high"))
+    json.dump({"name": name, "vec": vec.tolist(), "title": title,
+               "message": msg, "priority": prio},
+              open(os.path.join(SOUNDS_DIR, f"{name}.json"), "w"))
+    print(f"Saved fingerprint for '{name}'. ({len(load_templates())} sound(s) learned.)")
+
+
+def main():
+    if "--list" in sys.argv:
+        t = load_templates()
+        print("learned sounds:", ", ".join(t) or "(none)")
+        return
+    if "--learn" in sys.argv:
+        i = sys.argv.index("--learn")
+        if i + 1 >= len(sys.argv):
+            print("usage: --learn <name>  (e.g. targeted, depleted, warp, jump, dock)")
+            return
+        try:
+            learn(sys.argv[i + 1].lower())
+        except Exception as e:
+            print(f"learn failed (need numpy + soundcard, and audio playing): {e}")
+        return
+
+    cp = w.load_config()
+    try:
+        import numpy        # noqa: F401
+        import soundcard    # noqa: F401
+    except Exception as e:
+        print(f"[audio] deps missing ({e}); audio watcher idle. "
+              "update.py installs numpy+soundcard.")
+        return
+
+    templates = load_templates()
+    if not templates:
+        print("[audio] no learned sounds yet — run: python eve_audio_watcher.py --learn targeted")
+    print(f"[audio] started; {len(templates)} sound(s) (waits for !mining on)")
+    last = {}
+    cooldown = 8
+    announced = False
+    while True:
+        if not w.bot_mining(cp):
+            time.sleep(15)
+            continue
+        if not templates:
+            templates = load_templates()
+            time.sleep(10)
+            continue
+        if not announced:
+            w.heartbeat(cp, "audio")
+            announced = True
+        try:
+            chunk = _recorder(WINDOW_SECS)
+            if rms(chunk) < RMS_FLOOR:
+                continue
+            vec = fingerprint(chunk)
+            if vec is None:
+                continue
+            best, score = None, 0.0
+            for name, t in templates.items():
+                s = cosine(vec, t["vec"])
+                if s > score:
+                    best, score = name, s
+            if best and score >= MATCH_THRESHOLD and time.time() - last.get(best, 0) > cooldown:
+                last[best] = time.time()
+                t = templates[best]
+                w.notify(cp, t["title"], f"{t['message']} (heard ~{score:.0%})",
+                         priority=t.get("priority", "high"), tags="loudspeaker")
+                print(f"[audio] matched {best} ({score:.2f})")
+        except Exception as e:
+            print(f"[audio] error: {e}")
+            time.sleep(3)
+
+
+if __name__ == "__main__":
+    main()