David Gwilliam
e09bddb724
- Fix pre-existing lint errors in engine/ modules using ruff --unsafe-fixes - Add hk.pkl with pre-commit and pre-push hooks using ruff builtin - Configure hooks to use 'uv run' prefix for tool execution - Update mise.toml to include hk and pkl tools - All 73 tests pass fix: apply ruff auto-fixes and add hk git hooks - Fix pre-existing lint errors in engine/ modules using ruff --unsafe-fixes - Add hk.pkl with pre-commit and pre-push hooks using ruff builtin - Configure hooks to use 'uv run' prefix for tool execution - Update mise.toml to include hk and pkl tools - Use 'hk install --mise' for proper mise integration - All 73 tests pass
67 lines
1.6 KiB
Python
67 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
|