From 840e31ebac1d8813591fdc8f87d113bb57fa1343 Mon Sep 17 00:00:00 2001 From: brockdarnold Date: Sun, 14 Jun 2026 16:28:03 +0000 Subject: [PATCH] publish eve_audio_watcher.py --- eve_audio_watcher.py | 186 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 186 insertions(+) create mode 100644 eve_audio_watcher.py diff --git a/eve_audio_watcher.py b/eve_audio_watcher.py new file mode 100644 index 0000000..a0fe84c --- /dev/null +++ b/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 (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()