88 lines
3.2 KiB
Python
88 lines
3.2 KiB
Python
"""WASAPI loopback capture of the default speaker + detection of a reference sound in it."""
|
|
import threading
|
|
import warnings
|
|
|
|
import numpy as np
|
|
|
|
RATE = 48000
|
|
|
|
|
|
class LoopbackRecorder(threading.Thread):
|
|
def __init__(self, seconds, rate=RATE):
|
|
super().__init__(daemon=True)
|
|
self.seconds = seconds
|
|
self.rate = rate
|
|
self.chunks = []
|
|
self.stop_flag = False
|
|
self.error = None
|
|
|
|
def run(self):
|
|
try:
|
|
import soundcard as sc
|
|
sp = sc.default_speaker()
|
|
mic = sc.get_microphone(id=str(sp.name), include_loopback=True)
|
|
block = self.rate // 10
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore") # "data discontinuity" warnings
|
|
with mic.recorder(samplerate=self.rate, blocksize=block) as rec:
|
|
n = int(self.seconds * 10)
|
|
for _ in range(n):
|
|
if self.stop_flag:
|
|
break
|
|
self.chunks.append(rec.record(numframes=block))
|
|
except Exception as e: # noqa
|
|
self.error = repr(e)
|
|
|
|
def audio(self):
|
|
if not self.chunks:
|
|
return np.zeros(0, dtype=np.float32)
|
|
a = np.concatenate(self.chunks, axis=0)
|
|
return a.mean(axis=1).astype(np.float32) if a.ndim == 2 else a.astype(np.float32)
|
|
|
|
|
|
def resample(x, src_rate, dst_rate):
|
|
if src_rate == dst_rate:
|
|
return x.astype(np.float32)
|
|
n = int(round(len(x) * dst_rate / src_rate))
|
|
t = np.linspace(0, len(x) - 1, n)
|
|
return np.interp(t, np.arange(len(x)), x).astype(np.float32)
|
|
|
|
|
|
def detect(capture, ref, rate=RATE):
|
|
"""Max normalized cross-correlation of ref against every same-length window of capture.
|
|
Returns (score 0..1, time_offset_seconds, ref_rms_in_window)."""
|
|
if len(capture) < len(ref) or len(ref) == 0:
|
|
return 0.0, None, 0.0
|
|
ref = ref - ref.mean()
|
|
rn = np.linalg.norm(ref)
|
|
if rn == 0:
|
|
return 0.0, None, 0.0
|
|
n = 1 << int(np.ceil(np.log2(len(capture) + len(ref))))
|
|
xc = np.fft.irfft(np.fft.rfft(capture, n) * np.conj(np.fft.rfft(ref, n)), n)[:len(capture) - len(ref) + 1]
|
|
c2 = np.concatenate([[0.0], np.cumsum(capture.astype(np.float64) ** 2)])
|
|
win = np.sqrt(np.maximum(c2[len(ref):] - c2[:-len(ref)], 1e-12))
|
|
ncc = xc / (rn * win)
|
|
# ignore near-silent windows
|
|
ncc[win < 1e-3 * np.sqrt(len(ref))] = 0
|
|
i = int(np.argmax(np.abs(ncc)))
|
|
return float(abs(ncc[i])), i / rate, float(win[i] / np.sqrt(len(ref)))
|
|
|
|
|
|
def detect_chunks(capture, ref, rate=RATE, chunk_s=1.0, step_s=0.5):
|
|
"""Best NCC over 1 s chunks of the reference (robust to partial playback and resampler drift).
|
|
Returns (best score, chunk start in ref [s], position in capture [s])."""
|
|
c = int(chunk_s * rate)
|
|
s = int(step_s * rate)
|
|
best = (0.0, None, None)
|
|
if len(ref) <= c:
|
|
sc, t, _ = detect(capture, ref, rate)
|
|
return sc, 0.0, t
|
|
for i in range(0, len(ref) - c + 1, s):
|
|
seg = ref[i:i + c]
|
|
if np.sqrt(np.mean(seg ** 2)) < 1e-3:
|
|
continue
|
|
sc, t, _ = detect(capture, seg, rate)
|
|
if sc > best[0]:
|
|
best = (sc, i / rate, t)
|
|
return best
|