eve-watcher/eve_audio_watcher.py

#!/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()