David Gwilliam
bd4b146c02
- Fix import sorting (isort) across all engine modules - Fix SIM105 try-except-pass patterns (contextlib.suppress) - Fix nested with statements in tests - Fix unused loop variables Run 'uv run pytest' to verify tests still pass.
63 lines
1.6 KiB
Python
63 lines
1.6 KiB
Python
"""
|
|
Microphone input monitor — standalone, no internal dependencies.
|
|
Gracefully degrades if sounddevice/numpy are unavailable.
|
|
"""
|
|
|
|
import atexit
|
|
|
|
try:
|
|
import numpy as _np
|
|
import sounddevice as _sd
|
|
_HAS_MIC = True
|
|
except Exception:
|
|
_HAS_MIC = False
|
|
|
|
|
|
class MicMonitor:
|
|
"""Background mic stream that exposes current RMS dB level."""
|
|
|
|
def __init__(self, threshold_db=50):
|
|
self.threshold_db = threshold_db
|
|
self._db = -99.0
|
|
self._stream = None
|
|
|
|
@property
|
|
def available(self):
|
|
"""True if sounddevice is importable."""
|
|
return _HAS_MIC
|
|
|
|
@property
|
|
def db(self):
|
|
"""Current RMS dB level."""
|
|
return self._db
|
|
|
|
@property
|
|
def excess(self):
|
|
"""dB above threshold (clamped to 0)."""
|
|
return max(0.0, self._db - self.threshold_db)
|
|
|
|
def start(self):
|
|
"""Start background mic stream. Returns True on success, False/None otherwise."""
|
|
if not _HAS_MIC:
|
|
return None
|
|
def _cb(indata, frames, t, status):
|
|
rms = float(_np.sqrt(_np.mean(indata ** 2)))
|
|
self._db = 20 * _np.log10(rms) if rms > 0 else -99.0
|
|
try:
|
|
self._stream = _sd.InputStream(
|
|
callback=_cb, channels=1, samplerate=44100, blocksize=2048)
|
|
self._stream.start()
|
|
atexit.register(self.stop)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
def stop(self):
|
|
"""Stop the mic stream if running."""
|
|
if self._stream:
|
|
try:
|
|
self._stream.stop()
|
|
except Exception:
|
|
pass
|
|
self._stream = None
|