violawake_sdk

ViolaWake SDK — Open-source wake word detection + voice pipeline.

Public API surface:

WakeDetector — detect a wake word in an audio stream AsyncWakeDetector — async wrapper for asyncio-based applications DetectorConfig — advanced configuration (ensemble, adaptive, speaker, power) VADEngine — voice activity detection (WebRTC, Silero, RMS) TTSEngine — on-device text-to-speech (Kokoro-82M) STTEngine — speech-to-text (faster-whisper) VoicePipeline — bundled Wake→VAD→STT→TTS orchestration NoiseProfiler — noise-adaptive threshold adjustment PowerManager — battery-aware power management FusionStrategy — multi-model ensemble scoring list_models() — discover available wake word models list_voices() — discover available TTS voices

Quick start::

from violawake_sdk import WakeDetector

with WakeDetector(threshold=0.80) as detector:
    for chunk in detector.stream_mic():
        if detector.detect(chunk):
            print("Wake word detected!")
            break

See README.md or https://github.com/GeeIHadAGoodTime/ViolaWake for full documentation.

View Source

  1"""
  2ViolaWake SDK — Open-source wake word detection + voice pipeline.
  3
  4Public API surface:
  5    WakeDetector      — detect a wake word in an audio stream
  6    AsyncWakeDetector — async wrapper for asyncio-based applications
  7    DetectorConfig    — advanced configuration (ensemble, adaptive, speaker, power)
  8    VADEngine         — voice activity detection (WebRTC, Silero, RMS)
  9    TTSEngine         — on-device text-to-speech (Kokoro-82M)
 10    STTEngine         — speech-to-text (faster-whisper)
 11    VoicePipeline     — bundled Wake→VAD→STT→TTS orchestration
 12    NoiseProfiler     — noise-adaptive threshold adjustment
 13    PowerManager      — battery-aware power management
 14    FusionStrategy    — multi-model ensemble scoring
 15    list_models()     — discover available wake word models
 16    list_voices()     — discover available TTS voices
 17
 18Quick start::
 19
 20    from violawake_sdk import WakeDetector
 21
 22    with WakeDetector(threshold=0.80) as detector:
 23        for chunk in detector.stream_mic():
 24            if detector.detect(chunk):
 25                print("Wake word detected!")
 26                break
 27
 28See README.md or https://github.com/GeeIHadAGoodTime/ViolaWake for full documentation.
 29"""
 30
 31from __future__ import annotations
 32
 33# Read the installed distribution version so __version__ never drifts from
 34# pyproject.toml/PyPI. The fallback covers running from a source checkout
 35# without an editable install (rare in user code, common in CI before
 36# `pip install -e .`).
 37try:
 38    from importlib.metadata import PackageNotFoundError
 39    from importlib.metadata import version as _pkg_version
 40
 41    try:
 42        __version__ = _pkg_version("violawake")
 43    except PackageNotFoundError:
 44        __version__ = "0.0.0+source"
 45except ImportError:  # pragma: no cover - Python < 3.8 (unsupported)
 46    __version__ = "0.0.0+source"
 47
 48__author__ = "ViolaWake Contributors"
 49__license__ = "Apache-2.0"
 50
 51from violawake_sdk._exceptions import (
 52    AudioCaptureError,
 53    ModelLoadError,
 54    ModelNotFoundError,
 55    PipelineError,
 56    VADBackendError,
 57    ViolaWakeError,
 58)
 59from violawake_sdk.async_detector import AsyncWakeDetector
 60from violawake_sdk.confidence import ConfidenceLevel, ConfidenceResult
 61from violawake_sdk.ensemble import FusionStrategy
 62from violawake_sdk.noise_profiler import NoiseProfiler
 63from violawake_sdk.pipeline import AsyncVoicePipeline, VoicePipeline
 64from violawake_sdk.power_manager import PowerManager
 65from violawake_sdk.vad import VADEngine
 66from violawake_sdk.wake_detector import (
 67    DetectorConfig,
 68    WakeDecisionPolicy,
 69    WakeDetector,
 70    WakewordDetector,  # noqa: F401 — backward compat
 71    validate_audio_chunk,
 72)
 73
 74# Conditional imports for optional extras — provide helpful error on use if missing
 75
 76
 77def _make_missing_extra_class(name: str, extra: str):
 78    """Return a placeholder class that raises ImportError on instantiation."""
 79
 80    class _Missing:
 81        def __init__(self, *args, **kwargs):
 82            raise ImportError(
 83                f"{name} requires extra dependencies. "
 84                f'Install them with: pip install "violawake[{extra}]"'
 85            )
 86
 87    _Missing.__name__ = _Missing.__qualname__ = name
 88    return _Missing
 89
 90
 91try:
 92    from violawake_sdk.tts import TTSEngine
 93except ImportError:
 94    TTSEngine = _make_missing_extra_class("TTSEngine", "tts")  # type: ignore[assignment,misc]
 95
 96try:
 97    from violawake_sdk.stt import StreamingSTTEngine, STTEngine
 98except ImportError:
 99    STTEngine = _make_missing_extra_class("STTEngine", "stt")  # type: ignore[assignment,misc]
100    StreamingSTTEngine = _make_missing_extra_class("StreamingSTTEngine", "stt")  # type: ignore[assignment,misc]
101
102
103def list_models() -> list[dict[str, str]]:
104    """Return available wake word models with their descriptions.
105
106    Each entry is a dict with keys: ``name``, ``description``, ``version``.
107
108    Example::
109
110        >>> from violawake_sdk import list_models
111        >>> for m in list_models():
112        ...     print(f"{m['name']:20s} {m['description']}")
113    """
114    from violawake_sdk.models import MODEL_REGISTRY
115
116    seen: set[str] = set()
117    result: list[dict[str, str]] = []
118    for name, spec in MODEL_REGISTRY.items():
119        # Deduplicate aliases (e.g. "viola" -> "temporal_cnn")
120        if spec.name in seen:
121            continue
122        # Hide deprecated, package-managed, and non-wake-word models
123        if "DEPRECATED" in spec.description:
124            continue
125        if spec.name in ("oww_backbone", "kokoro_v1_0", "kokoro_voices_v1_0"):
126            continue
127        seen.add(spec.name)
128        result.append(
129            {
130                "name": name,
131                "description": spec.description,
132                "version": spec.version,
133            }
134        )
135    return result
136
137
138def list_voices() -> list[str]:
139    """Return available TTS voice names for use with ``TTSEngine``.
140
141    Requires the ``[tts]`` extra to be installed for actual synthesis,
142    but this function always works for discovery.
143
144    Example::
145
146        >>> from violawake_sdk import list_voices
147        >>> list_voices()
148        ['af_heart', 'af_bella', 'af_sarah', ...]
149    """
150    from violawake_sdk.tts import AVAILABLE_VOICES
151
152    return list(AVAILABLE_VOICES)
153
154
155__all__ = [
156    # Core detection
157    "DetectorConfig",
158    "WakeDetector",
159    "WakewordDetector",
160    "AsyncWakeDetector",
161    "WakeDecisionPolicy",
162    "validate_audio_chunk",
163    # Confidence & scoring
164    "ConfidenceResult",
165    "ConfidenceLevel",
166    "FusionStrategy",
167    # Advanced features
168    "NoiseProfiler",
169    "PowerManager",
170    # Pipeline components
171    "VADEngine",
172    "TTSEngine",
173    "STTEngine",
174    "StreamingSTTEngine",
175    "VoicePipeline",
176    "AsyncVoicePipeline",
177    # Exceptions
178    "ViolaWakeError",
179    "ModelNotFoundError",
180    "AudioCaptureError",
181    "ModelLoadError",
182    "PipelineError",
183    "VADBackendError",
184    # Discovery
185    "list_models",
186    "list_voices",
187    "__version__",
188]

@dataclass

class DetectorConfig: View Source

 71@dataclass
 72class DetectorConfig:
 73    """Advanced configuration for WakeDetector.
 74
 75    Basic usage needs no config -- just use ``WakeDetector(threshold=0.80)``.
 76    Use ``DetectorConfig`` to opt-in to advanced features without cluttering
 77    the constructor.
 78
 79    Example::
 80
 81        # Simple (80% of users):
 82        det = WakeDetector(model="temporal_cnn", threshold=0.80)
 83
 84        # Advanced (multi-model ensemble + adaptive threshold):
 85        det = WakeDetector(
 86            model="temporal_cnn",
 87            config=DetectorConfig(
 88                adaptive_threshold=True,
 89                confirm_count=3,
 90            ),
 91        )
 92
 93    Attributes:
 94        models: Additional model paths for multi-model ensemble (K3).
 95        fusion_strategy: Score fusion strategy for ensemble (K3).
 96        fusion_weights: Per-model weights for weighted_average fusion (K3).
 97        adaptive_threshold: Enable dynamic threshold based on noise (K4).
 98        noise_profiler: Custom NoiseProfiler instance (K4).
 99        speaker_verify_fn: Post-detection speaker verification callback (K5).
100        power_manager: Power management controller for duty cycling (K7).
101        confirm_count: Consecutive above-threshold scores required (K2).
102        score_history_size: Number of recent scores to retain (K2).
103    """
104
105    # K3: Multi-model ensemble
106    models: list[str] | None = None
107    fusion_strategy: FusionStrategy | str = FusionStrategy.AVERAGE
108    fusion_weights: list[float] | None = None
109
110    # K4: Adaptive threshold
111    adaptive_threshold: bool = False
112    noise_profiler: NoiseProfiler | None = None
113
114    # K5: Speaker verification
115    speaker_verify_fn: Callable[..., bool] | None = None
116
117    # K7: Power management
118    power_manager: PowerManager | None = None
119
120    # K2: Confidence tracking
121    confirm_count: int = 1
122    score_history_size: int = 50
123
124    def build(self, model: str = "temporal_cnn", **kwargs: Any) -> WakeDetector:
125        """Build a WakeDetector from this config.
126
127        Convenience method that passes ``self`` as the ``config=`` argument.
128
129        Args:
130            model: Model name or path (default: ``"temporal_cnn"``).
131            **kwargs: Additional WakeDetector constructor arguments
132                (threshold, cooldown_s, etc.).
133
134        Returns:
135            Configured WakeDetector instance.
136        """
137        return WakeDetector(model=model, config=self, **kwargs)

Advanced configuration for WakeDetector.

Basic usage needs no config -- just use WakeDetector(threshold=0.80). Use DetectorConfig to opt-in to advanced features without cluttering the constructor.

Example::

# Simple (80% of users):
det = WakeDetector(model="temporal_cnn", threshold=0.80)

# Advanced (multi-model ensemble + adaptive threshold):
det = WakeDetector(
    model="temporal_cnn",
    config=DetectorConfig(
        adaptive_threshold=True,
        confirm_count=3,
    ),
)

Attributes:

models: Additional model paths for multi-model ensemble (K3).
fusion_strategy: Score fusion strategy for ensemble (K3).
fusion_weights: Per-model weights for weighted_average fusion (K3).
adaptive_threshold: Enable dynamic threshold based on noise (K4).
noise_profiler: Custom NoiseProfiler instance (K4).
speaker_verify_fn: Post-detection speaker verification callback (K5).
power_manager: Power management controller for duty cycling (K7).
confirm_count: Consecutive above-threshold scores required (K2).
score_history_size: Number of recent scores to retain (K2).

DetectorConfig( models: list[str] | None = None, fusion_strategy: FusionStrategy | str = <FusionStrategy.AVERAGE: 'average'>, fusion_weights: list[float] | None = None, adaptive_threshold: bool = False, noise_profiler: NoiseProfiler | None = None, speaker_verify_fn: Callable[..., bool] | None = None, power_manager: PowerManager | None = None, confirm_count: int = 1, score_history_size: int = 50)

models: list[str] | None = None

fusion_strategy: FusionStrategy | str = <FusionStrategy.AVERAGE: 'average'>

fusion_weights: list[float] | None = None

adaptive_threshold: bool = False

noise_profiler: NoiseProfiler | None = None

speaker_verify_fn: Callable[..., bool] | None = None

power_manager: PowerManager | None = None

confirm_count: int = 1

score_history_size: int = 50

def build( self, model: str = 'temporal_cnn', **kwargs: Any) -> WakeDetector: View Source

124    def build(self, model: str = "temporal_cnn", **kwargs: Any) -> WakeDetector:
125        """Build a WakeDetector from this config.
126
127        Convenience method that passes ``self`` as the ``config=`` argument.
128
129        Args:
130            model: Model name or path (default: ``"temporal_cnn"``).
131            **kwargs: Additional WakeDetector constructor arguments
132                (threshold, cooldown_s, etc.).
133
134        Returns:
135            Configured WakeDetector instance.
136        """
137        return WakeDetector(model=model, config=self, **kwargs)

Build a WakeDetector from this config.

Convenience method that passes self as the config= argument.

Arguments:

model: Model name or path (default: "temporal_cnn").
**kwargs: Additional WakeDetector constructor arguments (threshold, cooldown_s, etc.).

Returns:

Configured WakeDetector instance.

class AsyncWakeDetector: View Source

 38class AsyncWakeDetector:
 39    """Async wrapper around ``WakeDetector`` for asyncio-based applications.
 40
 41    All CPU-bound inference is dispatched to a background thread via
 42    ``loop.run_in_executor``. The wrapper is fully transparent -- it
 43    accepts the same constructor arguments and exposes the same methods
 44    as ``WakeDetector``, but with ``async`` signatures.
 45
 46    Args:
 47        **kwargs: Forwarded to ``WakeDetector.__init__``.
 48    """
 49
 50    def __init__(self, **kwargs: Any) -> None:
 51        self._detector = WakeDetector(**kwargs)
 52        self._executor: ThreadPoolExecutor | None = None
 53
 54    def _get_executor(self) -> ThreadPoolExecutor:
 55        if self._executor is None:
 56            self._executor = ThreadPoolExecutor(max_workers=1)
 57        return self._executor
 58
 59    async def __aenter__(self) -> AsyncWakeDetector:
 60        """Enter async context manager."""
 61        return self
 62
 63    async def __aexit__(
 64        self, exc_type: type | None, exc_val: BaseException | None, exc_tb: object
 65    ) -> None:
 66        """Exit async context manager, shutting down the executor."""
 67        self.close()
 68
 69    async def detect(
 70        self,
 71        audio_frame: bytes | np.ndarray,
 72        is_playing: bool = False,
 73    ) -> bool:
 74        """Async version of ``WakeDetector.detect``."""
 75        loop = asyncio.get_running_loop()
 76        return await loop.run_in_executor(
 77            self._get_executor(),
 78            lambda: self._detector.detect(audio_frame, is_playing),
 79        )
 80
 81    async def process(
 82        self,
 83        audio_frame: bytes | np.ndarray,
 84    ) -> float:
 85        """Async version of ``WakeDetector.process``."""
 86        loop = asyncio.get_running_loop()
 87        return await loop.run_in_executor(
 88            self._get_executor(),
 89            lambda: self._detector.process(audio_frame),
 90        )
 91
 92    async def stream(
 93        self,
 94        source: AsyncIterator[bytes | np.ndarray],
 95    ) -> AsyncIterator[bool]:
 96        """Async generator that yields detection results from an async audio source.
 97
 98        Usage::
 99
100            async for detected in detector.stream(audio_source):
101                if detected:
102                    print("Wake word!")
103
104        Args:
105            source: An async iterator yielding audio frames.
106
107        Yields:
108            Boolean detection result for each frame.
109        """
110        async for frame in source:
111            yield await self.detect(frame)
112
113    def reset_cooldown(self) -> None:
114        """Reset the cooldown window (delegates to WakeDetector public API)."""
115        self._detector.reset_cooldown()
116
117    @property
118    def threshold(self) -> float:
119        """Current detection threshold."""
120        return self._detector.threshold
121
122    def get_confidence(self) -> ConfidenceResult:
123        """Return confidence assessment of the current detection state (K2)."""
124        return self._detector.get_confidence()
125
126    @property
127    def last_scores(self) -> tuple[float, ...]:
128        """Return the recent score history (most recent last)."""
129        return self._detector.last_scores
130
131    async def aclose(self) -> None:
132        """Async shutdown — shuts down executor in a thread to avoid blocking."""
133        loop = asyncio.get_running_loop()
134        if self._executor is not None:
135            executor = self._executor
136            self._executor = None
137            await loop.run_in_executor(None, lambda: executor.shutdown(wait=True))
138        self._detector.close()
139
140    def close(self) -> None:
141        """Shut down the background executor and release detector resources.
142
143        Safe to call multiple times.  For async contexts prefer ``aclose()``.
144        """
145        if self._executor is not None:
146            self._executor.shutdown(wait=True)
147            self._executor = None
148        self._detector.close()
149
150    def __del__(self) -> None:
151        # Use wait=False in __del__ to avoid blocking the GC/finalizer thread.
152        # The explicit close() method still uses wait=True for graceful shutdown.
153        if self._executor is not None:
154            self._executor.shutdown(wait=False)
155            self._executor = None

Async wrapper around WakeDetector for asyncio-based applications.

All CPU-bound inference is dispatched to a background thread via loop.run_in_executor. The wrapper is fully transparent -- it accepts the same constructor arguments and exposes the same methods as WakeDetector, but with async signatures.

Arguments:

**kwargs: Forwarded to WakeDetector.__init__.

AsyncWakeDetector(**kwargs: Any) View Source

50    def __init__(self, **kwargs: Any) -> None:
51        self._detector = WakeDetector(**kwargs)
52        self._executor: ThreadPoolExecutor | None = None

async def detect( self, audio_frame: bytes | numpy.ndarray, is_playing: bool = False) -> bool: View Source

69    async def detect(
70        self,
71        audio_frame: bytes | np.ndarray,
72        is_playing: bool = False,
73    ) -> bool:
74        """Async version of ``WakeDetector.detect``."""
75        loop = asyncio.get_running_loop()
76        return await loop.run_in_executor(
77            self._get_executor(),
78            lambda: self._detector.detect(audio_frame, is_playing),
79        )

Async version of WakeDetector.detect.

async def process(self, audio_frame: bytes | numpy.ndarray) -> float: View Source

81    async def process(
82        self,
83        audio_frame: bytes | np.ndarray,
84    ) -> float:
85        """Async version of ``WakeDetector.process``."""
86        loop = asyncio.get_running_loop()
87        return await loop.run_in_executor(
88            self._get_executor(),
89            lambda: self._detector.process(audio_frame),
90        )

Async version of WakeDetector.process.

async def stream( self, source: AsyncIterator[bytes | numpy.ndarray]) -> AsyncIterator[bool]: View Source

 92    async def stream(
 93        self,
 94        source: AsyncIterator[bytes | np.ndarray],
 95    ) -> AsyncIterator[bool]:
 96        """Async generator that yields detection results from an async audio source.
 97
 98        Usage::
 99
100            async for detected in detector.stream(audio_source):
101                if detected:
102                    print("Wake word!")
103
104        Args:
105            source: An async iterator yielding audio frames.
106
107        Yields:
108            Boolean detection result for each frame.
109        """
110        async for frame in source:
111            yield await self.detect(frame)

Async generator that yields detection results from an async audio source.

Usage::

async for detected in detector.stream(audio_source):
    if detected:
        print("Wake word!")

Arguments:

source: An async iterator yielding audio frames.

Yields:

Boolean detection result for each frame.

def reset_cooldown(self) -> None: View Source

113    def reset_cooldown(self) -> None:
114        """Reset the cooldown window (delegates to WakeDetector public API)."""
115        self._detector.reset_cooldown()

Reset the cooldown window (delegates to WakeDetector public API).

threshold: float View Source

117    @property
118    def threshold(self) -> float:
119        """Current detection threshold."""
120        return self._detector.threshold

Current detection threshold.

def get_confidence(self) -> ConfidenceResult: View Source

122    def get_confidence(self) -> ConfidenceResult:
123        """Return confidence assessment of the current detection state (K2)."""
124        return self._detector.get_confidence()

Return confidence assessment of the current detection state (K2).

last_scores: tuple[float, ...] View Source

126    @property
127    def last_scores(self) -> tuple[float, ...]:
128        """Return the recent score history (most recent last)."""
129        return self._detector.last_scores

Return the recent score history (most recent last).

async def aclose(self) -> None: View Source

131    async def aclose(self) -> None:
132        """Async shutdown — shuts down executor in a thread to avoid blocking."""
133        loop = asyncio.get_running_loop()
134        if self._executor is not None:
135            executor = self._executor
136            self._executor = None
137            await loop.run_in_executor(None, lambda: executor.shutdown(wait=True))
138        self._detector.close()

Async shutdown — shuts down executor in a thread to avoid blocking.

def close(self) -> None: View Source

140    def close(self) -> None:
141        """Shut down the background executor and release detector resources.
142
143        Safe to call multiple times.  For async contexts prefer ``aclose()``.
144        """
145        if self._executor is not None:
146            self._executor.shutdown(wait=True)
147            self._executor = None
148        self._detector.close()

Shut down the background executor and release detector resources.

Safe to call multiple times. For async contexts prefer aclose().

class WakeDecisionPolicy: View Source

218class WakeDecisionPolicy:
219    """4-gate core decision pipeline (RMS floor, threshold, cooldown, playback suppression).
220
221    Extended by WakeDetector with optional confirmation (K2), adaptive
222    threshold (K4), and speaker verification (K5).
223
224    Gate 1: Zero-input guard -- skip if RMS < 1.0 (silence / DC offset artifact)
225    Gate 2: Score threshold -- skip if model score < threshold
226    Gate 3: Cooldown -- ignore events within cooldown_s of last detection
227    Gate 4: Listening gate -- suppress during active playback (optional)
228    """
229
230    def __init__(
231        self,
232        threshold: float = DEFAULT_THRESHOLD,
233        cooldown_s: float = DEFAULT_COOLDOWN_S,
234        rms_floor: float = 1.0,
235    ) -> None:
236        if not 0.0 <= threshold <= 1.0:
237            raise ValueError(f"threshold must be in [0.0, 1.0], got {threshold!r}")
238
239        self.threshold = threshold
240        self.cooldown_s = cooldown_s
241        self.rms_floor = rms_floor
242        self._last_detection: float = 0.0
243
244    def evaluate(
245        self,
246        score: float,
247        rms: float = 100.0,
248        is_playing: bool = False,
249    ) -> bool:
250        """Evaluate whether a wake word event should be triggered."""
251        if rms < self.rms_floor:
252            logger.debug("Gate 1 reject: RMS %.1f below floor %.1f", rms, self.rms_floor)
253            return False
254        if score < self.threshold:
255            return False
256        now = time.monotonic()
257        if now - self._last_detection < self.cooldown_s:
258            logger.debug(
259                "Gate 3 reject: cooldown active (%.1fs remaining)",
260                self.cooldown_s - (now - self._last_detection),
261            )
262            return False
263        if is_playing:
264            logger.debug("Gate 4 reject: playback active")
265            return False
266        self._last_detection = now
267        logger.info("Wake word detected! score=%.3f", score)
268        return True
269
270    def reset_cooldown(self) -> None:
271        """Reset the cooldown window (useful for testing)."""
272        self._last_detection = 0.0

4-gate core decision pipeline (RMS floor, threshold, cooldown, playback suppression).

Extended by WakeDetector with optional confirmation (K2), adaptive threshold (K4), and speaker verification (K5).

Gate 1: Zero-input guard -- skip if RMS < 1.0 (silence / DC offset artifact) Gate 2: Score threshold -- skip if model score < threshold Gate 3: Cooldown -- ignore events within cooldown_s of last detection Gate 4: Listening gate -- suppress during active playback (optional)

WakeDecisionPolicy( threshold: float = 0.8, cooldown_s: float = 2.0, rms_floor: float = 1.0) View Source

230    def __init__(
231        self,
232        threshold: float = DEFAULT_THRESHOLD,
233        cooldown_s: float = DEFAULT_COOLDOWN_S,
234        rms_floor: float = 1.0,
235    ) -> None:
236        if not 0.0 <= threshold <= 1.0:
237            raise ValueError(f"threshold must be in [0.0, 1.0], got {threshold!r}")
238
239        self.threshold = threshold
240        self.cooldown_s = cooldown_s
241        self.rms_floor = rms_floor
242        self._last_detection: float = 0.0

threshold

cooldown_s

rms_floor

def evaluate(self, score: float, rms: float = 100.0, is_playing: bool = False) -> bool: View Source

244    def evaluate(
245        self,
246        score: float,
247        rms: float = 100.0,
248        is_playing: bool = False,
249    ) -> bool:
250        """Evaluate whether a wake word event should be triggered."""
251        if rms < self.rms_floor:
252            logger.debug("Gate 1 reject: RMS %.1f below floor %.1f", rms, self.rms_floor)
253            return False
254        if score < self.threshold:
255            return False
256        now = time.monotonic()
257        if now - self._last_detection < self.cooldown_s:
258            logger.debug(
259                "Gate 3 reject: cooldown active (%.1fs remaining)",
260                self.cooldown_s - (now - self._last_detection),
261            )
262            return False
263        if is_playing:
264            logger.debug("Gate 4 reject: playback active")
265            return False
266        self._last_detection = now
267        logger.info("Wake word detected! score=%.3f", score)
268        return True

Evaluate whether a wake word event should be triggered.

def reset_cooldown(self) -> None: View Source

270    def reset_cooldown(self) -> None:
271        """Reset the cooldown window (useful for testing)."""
272        self._last_detection = 0.0

Reset the cooldown window (useful for testing).

def validate_audio_chunk(data: bytes | numpy.ndarray) -> numpy.ndarray: View Source

162def validate_audio_chunk(data: bytes | np.ndarray) -> np.ndarray:
163    """Validate and normalize an audio chunk for use with WakeDetector.
164
165    Accepts bytes (int16 PCM) or numpy arrays (int16, float32, float64).
166    Returns a float32 numpy array suitable for processing.
167
168    Args:
169        data: Audio chunk as bytes (int16 little-endian PCM) or numpy array.
170
171    Returns:
172        Validated float32 numpy array.
173
174    Raises:
175        TypeError: If data is not bytes or ndarray.
176        ValueError: If data is empty, has invalid dtype, contains only
177            non-finite values, or exceeds the maximum chunk size.
178    """
179    if isinstance(data, bytes):
180        if len(data) == 0:
181            raise ValueError("Audio chunk is empty (0 bytes)")
182        if len(data) % 2 != 0:
183            raise ValueError(
184                f"Audio bytes length must be even (int16 = 2 bytes/sample), got {len(data)}"
185            )
186        pcm = np.frombuffer(data, dtype=np.int16).astype(np.float32)
187    elif isinstance(data, np.ndarray):
188        if data.size == 0:
189            raise ValueError("Audio chunk is empty (0 samples)")
190        if data.ndim != 1:
191            raise ValueError(
192                f"Audio chunk must be 1-D, got {data.ndim}-D array with shape {data.shape}"
193            )
194        original_dtype = data.dtype
195        if original_dtype not in _ALLOWED_AUDIO_DTYPES:
196            raise ValueError(
197                f"Audio chunk dtype must be one of {[str(d) for d in _ALLOWED_AUDIO_DTYPES]}, "
198                f"got {original_dtype}"
199            )
200        pcm = data.astype(np.float32)
201        # Check the pre-cast dtype here; after astype(float32) every ndarray looks floating,
202        # which made the old post-cast dtype branch in the process() path effectively dead.
203        if original_dtype != np.int16 and not np.all(np.isfinite(pcm)):
204            pcm = np.where(np.isfinite(pcm), pcm, 0.0).astype(np.float32, copy=False)
205            logger.warning("Audio chunk contained non-finite values (NaN/inf); replaced with 0")
206    else:
207        raise TypeError(f"Audio chunk must be bytes or numpy ndarray, got {type(data).__name__}")
208
209    if len(pcm) > _MAX_CHUNK_SAMPLES:
210        raise ValueError(
211            f"Audio chunk too large: {len(pcm)} samples "
212            f"(max {_MAX_CHUNK_SAMPLES} = {_MAX_CHUNK_SAMPLES // SAMPLE_RATE}s at {SAMPLE_RATE}Hz)"
213        )
214
215    return pcm

Validate and normalize an audio chunk for use with WakeDetector.

Accepts bytes (int16 PCM) or numpy arrays (int16, float32, float64). Returns a float32 numpy array suitable for processing.

Arguments:

data: Audio chunk as bytes (int16 little-endian PCM) or numpy array.

Returns:

Validated float32 numpy array.

Raises:

TypeError: If data is not bytes or ndarray.
ValueError: If data is empty, has invalid dtype, contains only non-finite values, or exceeds the maximum chunk size.

@dataclass(frozen=True)

class ConfidenceResult: View Source

28@dataclass(frozen=True)
29class ConfidenceResult:
30    """Result from get_confidence() with full detection context.
31
32    Attributes:
33        raw_score: The most recent MLP/CNN output score in [0.0, 1.0].
34        confirm_count: Number of consecutive above-threshold scores in the
35            current multi-window confirmation sequence.
36        confirm_required: Total consecutive scores required for detection.
37        confidence: Classified confidence level.
38        score_history: Recent score history (most recent last).
39    """
40
41    raw_score: float
42    confirm_count: int
43    confirm_required: int
44    confidence: ConfidenceLevel
45    score_history: tuple[float, ...]

Result from get_confidence() with full detection context.

Attributes:

raw_score: The most recent MLP/CNN output score in [0.0, 1.0].
confirm_count: Number of consecutive above-threshold scores in the current multi-window confirmation sequence.
confirm_required: Total consecutive scores required for detection.
confidence: Classified confidence level.
score_history: Recent score history (most recent last).

ConfidenceResult( raw_score: float, confirm_count: int, confirm_required: int, confidence: ConfidenceLevel, score_history: tuple[float, ...])

raw_score: float

confirm_count: int

confirm_required: int

confidence: ConfidenceLevel

score_history: tuple[float, ...]

class ConfidenceLevel(builtins.str, enum.Enum): View Source

19class ConfidenceLevel(str, Enum):
20    """Confidence classification for wake word detection."""
21
22    LOW = "LOW"
23    MEDIUM = "MEDIUM"
24    HIGH = "HIGH"
25    CERTAIN = "CERTAIN"

Confidence classification for wake word detection.

LOW = <ConfidenceLevel.LOW: 'LOW'>

MEDIUM = <ConfidenceLevel.MEDIUM: 'MEDIUM'>

HIGH = <ConfidenceLevel.HIGH: 'HIGH'>

CERTAIN = <ConfidenceLevel.CERTAIN: 'CERTAIN'>

class FusionStrategy(builtins.str, enum.Enum): View Source

27class FusionStrategy(str, Enum):
28    """Score fusion strategy for multi-model ensemble."""
29
30    AVERAGE = "average"
31    MAX = "max"
32    VOTING = "voting"
33    WEIGHTED_AVERAGE = "weighted_average"

Score fusion strategy for multi-model ensemble.

AVERAGE = <FusionStrategy.AVERAGE: 'average'>

MAX = <FusionStrategy.MAX: 'max'>

VOTING = <FusionStrategy.VOTING: 'voting'>

WEIGHTED_AVERAGE = <FusionStrategy.WEIGHTED_AVERAGE: 'weighted_average'>

class NoiseProfiler: View Source

 47class NoiseProfiler:
 48    """Estimates ambient noise and adjusts detection threshold.
 49
 50    The profiler maintains a rolling window of RMS energy measurements.
 51    The noise floor is estimated as the 10th percentile of recent RMS values
 52    (capturing the quietest frames, which are likely ambient noise).
 53
 54    Threshold adjustment logic:
 55    - High SNR (signal clearly above noise): lower threshold slightly to
 56      improve sensitivity.
 57    - Low SNR (signal barely above noise): raise threshold to reduce
 58      false alarms from noise bursts.
 59    - The adjusted threshold is always clamped to
 60      ``[min_threshold, max_threshold]``.
 61
 62    Adaptive threshold bounds:
 63    - ``min_threshold`` defaults to ``0.60``. In very quiet environments, or
 64      when the current frame is far above the estimated noise floor, the
 65      threshold may be lowered to improve sensitivity, but it will never be
 66      reduced below this floor.
 67    - ``max_threshold`` defaults to ``0.95``. In very noisy environments, or
 68      when the current frame is close to the estimated noise floor, the
 69      threshold may be raised to suppress false accepts, but it will never be
 70      increased above this ceiling.
 71    - Before enough history is collected (fewer than 10 RMS frames), the
 72      profiler returns ``base_threshold`` with no adaptation.
 73
 74    Args:
 75        base_threshold: The default detection threshold (e.g. 0.80).
 76        noise_window_s: Seconds of audio history for noise estimation.
 77        min_threshold: Floor for adaptive threshold. Default 0.60.
 78        max_threshold: Ceiling for adaptive threshold. Default 0.95.
 79        snr_boost_db: SNR above this value enables threshold lowering.
 80        snr_penalty_db: SNR below this value enables threshold raising.
 81        frames_per_second: Expected audio frames per second (default 50 for 20ms).
 82    """
 83
 84    def __init__(
 85        self,
 86        base_threshold: float = 0.80,
 87        noise_window_s: float = DEFAULT_NOISE_WINDOW_S,
 88        min_threshold: float = DEFAULT_MIN_THRESHOLD,
 89        max_threshold: float = DEFAULT_MAX_THRESHOLD,
 90        snr_boost_db: float = DEFAULT_SNR_BOOST_DB,
 91        snr_penalty_db: float = DEFAULT_SNR_PENALTY_DB,
 92        frames_per_second: float = 50.0,
 93    ) -> None:
 94        self._base_threshold = base_threshold
 95        self._min_threshold = min_threshold
 96        self._max_threshold = max_threshold
 97        self._snr_boost_db = snr_boost_db
 98        self._snr_penalty_db = snr_penalty_db
 99
100        window_frames = max(1, int(noise_window_s * frames_per_second))
101        self._rms_history: deque[float] = deque(maxlen=window_frames)
102        self._current_rms: float = 0.0
103        self._noise_floor_rms: float = 0.0
104
105    @property
106    def base_threshold(self) -> float:
107        """The unadjusted detection threshold."""
108        return self._base_threshold
109
110    @property
111    def noise_floor(self) -> float:
112        """Current estimated noise floor RMS."""
113        return self._noise_floor_rms
114
115    def update(self, audio_frame: np.ndarray) -> float:
116        """Update noise estimate with a new audio frame and return adjusted threshold.
117
118        The audio should be float32 values. Both normalized [-1,1] and int16-range
119        float32 are accepted; the profiler works on relative ratios so absolute
120        scale doesn't matter.
121
122        Args:
123            audio_frame: 1-D float32 audio samples (any length).
124
125        Returns:
126            The adaptively adjusted detection threshold.
127        """
128        rms = float(np.sqrt(np.mean(audio_frame.astype(np.float64) ** 2)))
129        self._current_rms = rms
130        self._rms_history.append(rms)
131
132        # Estimate noise floor as 10th percentile of recent RMS values
133        if len(self._rms_history) >= 10:
134            sorted_rms = sorted(self._rms_history)
135            idx = max(0, int(len(sorted_rms) * 0.10))
136            self._noise_floor_rms = sorted_rms[idx]
137        elif self._rms_history:
138            self._noise_floor_rms = min(self._rms_history)
139        else:
140            self._noise_floor_rms = 0.0
141
142        return self._compute_adjusted_threshold()
143
144    def _compute_adjusted_threshold(self) -> float:
145        """Compute threshold adjustment based on current SNR estimate."""
146        snr_db = self._estimate_snr_db()
147
148        # No adjustment if we don't have enough data
149        if len(self._rms_history) < 10:
150            return self._base_threshold
151
152        if snr_db > self._snr_boost_db:
153            # High SNR: lower threshold proportionally (max 0.10 reduction)
154            excess = snr_db - self._snr_boost_db
155            reduction = min(0.10, excess * 0.01)
156            adjusted = self._base_threshold - reduction
157        elif snr_db < self._snr_penalty_db:
158            # Low SNR: raise threshold proportionally (max 0.10 increase)
159            deficit = self._snr_penalty_db - snr_db
160            increase = min(0.10, deficit * 0.02)
161            adjusted = self._base_threshold + increase
162        else:
163            adjusted = self._base_threshold
164
165        return max(self._min_threshold, min(self._max_threshold, adjusted))
166
167    def _estimate_snr_db(self) -> float:
168        """Estimate signal-to-noise ratio in decibels.
169
170        Uses the current frame RMS as signal and the noise floor as noise.
171        Returns 0.0 if noise floor is effectively zero.
172        """
173        if self._noise_floor_rms < 1e-10:
174            # No noise estimate yet — return a neutral value
175            return self._snr_boost_db  # neutral, no adjustment
176        if self._current_rms < 1e-10:
177            return 0.0
178
179        ratio = self._current_rms / self._noise_floor_rms
180        return 20.0 * math.log10(max(ratio, 1e-10))
181
182    def get_profile(self) -> NoiseProfile:
183        """Return a snapshot of the current noise state.
184
185        Returns:
186            NoiseProfile with noise floor, signal RMS, SNR, and adjusted threshold.
187        """
188        snr_db = self._estimate_snr_db()
189        adjusted = self._compute_adjusted_threshold()
190
191        return NoiseProfile(
192            noise_rms=self._noise_floor_rms,
193            signal_rms=self._current_rms,
194            snr_db=snr_db,
195            adjusted_threshold=adjusted,
196            base_threshold=self._base_threshold,
197        )
198
199    def reset(self) -> None:
200        """Clear noise history and reset estimates."""
201        self._rms_history.clear()
202        self._current_rms = 0.0
203        self._noise_floor_rms = 0.0

Estimates ambient noise and adjusts detection threshold.

The profiler maintains a rolling window of RMS energy measurements. The noise floor is estimated as the 10th percentile of recent RMS values (capturing the quietest frames, which are likely ambient noise).

Threshold adjustment logic:

High SNR (signal clearly above noise): lower threshold slightly to improve sensitivity.
Low SNR (signal barely above noise): raise threshold to reduce false alarms from noise bursts.
The adjusted threshold is always clamped to [min_threshold, max_threshold].

Adaptive threshold bounds:

min_threshold defaults to 0.60. In very quiet environments, or when the current frame is far above the estimated noise floor, the threshold may be lowered to improve sensitivity, but it will never be reduced below this floor.
max_threshold defaults to 0.95. In very noisy environments, or when the current frame is close to the estimated noise floor, the threshold may be raised to suppress false accepts, but it will never be increased above this ceiling.
Before enough history is collected (fewer than 10 RMS frames), the profiler returns base_threshold with no adaptation.

Arguments:

base_threshold: The default detection threshold (e.g. 0.80).
noise_window_s: Seconds of audio history for noise estimation.
min_threshold: Floor for adaptive threshold. Default 0.60.
max_threshold: Ceiling for adaptive threshold. Default 0.95.
snr_boost_db: SNR above this value enables threshold lowering.
snr_penalty_db: SNR below this value enables threshold raising.
frames_per_second: Expected audio frames per second (default 50 for 20ms).

NoiseProfiler( base_threshold: float = 0.8, noise_window_s: float = 5.0, min_threshold: float = 0.6, max_threshold: float = 0.95, snr_boost_db: float = 6.0, snr_penalty_db: float = 3.0, frames_per_second: float = 50.0) View Source

 84    def __init__(
 85        self,
 86        base_threshold: float = 0.80,
 87        noise_window_s: float = DEFAULT_NOISE_WINDOW_S,
 88        min_threshold: float = DEFAULT_MIN_THRESHOLD,
 89        max_threshold: float = DEFAULT_MAX_THRESHOLD,
 90        snr_boost_db: float = DEFAULT_SNR_BOOST_DB,
 91        snr_penalty_db: float = DEFAULT_SNR_PENALTY_DB,
 92        frames_per_second: float = 50.0,
 93    ) -> None:
 94        self._base_threshold = base_threshold
 95        self._min_threshold = min_threshold
 96        self._max_threshold = max_threshold
 97        self._snr_boost_db = snr_boost_db
 98        self._snr_penalty_db = snr_penalty_db
 99
100        window_frames = max(1, int(noise_window_s * frames_per_second))
101        self._rms_history: deque[float] = deque(maxlen=window_frames)
102        self._current_rms: float = 0.0
103        self._noise_floor_rms: float = 0.0

base_threshold: float View Source

105    @property
106    def base_threshold(self) -> float:
107        """The unadjusted detection threshold."""
108        return self._base_threshold

The unadjusted detection threshold.

noise_floor: float View Source

110    @property
111    def noise_floor(self) -> float:
112        """Current estimated noise floor RMS."""
113        return self._noise_floor_rms

Current estimated noise floor RMS.

def update(self, audio_frame: numpy.ndarray) -> float: View Source

115    def update(self, audio_frame: np.ndarray) -> float:
116        """Update noise estimate with a new audio frame and return adjusted threshold.
117
118        The audio should be float32 values. Both normalized [-1,1] and int16-range
119        float32 are accepted; the profiler works on relative ratios so absolute
120        scale doesn't matter.
121
122        Args:
123            audio_frame: 1-D float32 audio samples (any length).
124
125        Returns:
126            The adaptively adjusted detection threshold.
127        """
128        rms = float(np.sqrt(np.mean(audio_frame.astype(np.float64) ** 2)))
129        self._current_rms = rms
130        self._rms_history.append(rms)
131
132        # Estimate noise floor as 10th percentile of recent RMS values
133        if len(self._rms_history) >= 10:
134            sorted_rms = sorted(self._rms_history)
135            idx = max(0, int(len(sorted_rms) * 0.10))
136            self._noise_floor_rms = sorted_rms[idx]
137        elif self._rms_history:
138            self._noise_floor_rms = min(self._rms_history)
139        else:
140            self._noise_floor_rms = 0.0
141
142        return self._compute_adjusted_threshold()

Update noise estimate with a new audio frame and return adjusted threshold.

The audio should be float32 values. Both normalized [-1,1] and int16-range float32 are accepted; the profiler works on relative ratios so absolute scale doesn't matter.

Arguments:

audio_frame: 1-D float32 audio samples (any length).

Returns:

The adaptively adjusted detection threshold.

def get_profile(self) -> violawake_sdk.noise_profiler.NoiseProfile: View Source

182    def get_profile(self) -> NoiseProfile:
183        """Return a snapshot of the current noise state.
184
185        Returns:
186            NoiseProfile with noise floor, signal RMS, SNR, and adjusted threshold.
187        """
188        snr_db = self._estimate_snr_db()
189        adjusted = self._compute_adjusted_threshold()
190
191        return NoiseProfile(
192            noise_rms=self._noise_floor_rms,
193            signal_rms=self._current_rms,
194            snr_db=snr_db,
195            adjusted_threshold=adjusted,
196            base_threshold=self._base_threshold,
197        )

Return a snapshot of the current noise state.

Returns:

NoiseProfile with noise floor, signal RMS, SNR, and adjusted threshold.

def reset(self) -> None: View Source

199    def reset(self) -> None:
200        """Clear noise history and reset estimates."""
201        self._rms_history.clear()
202        self._current_rms = 0.0
203        self._noise_floor_rms = 0.0

Clear noise history and reset estimates.

class PowerManager: View Source

 99class PowerManager:
100    """Energy-aware inference controller.
101
102    Reduces inference frequency based on battery level, silence detection,
103    and explicit duty cycling configuration.
104
105    Modes of power saving:
106    1. **Duty cycling**: Process every Nth frame when idle (no recent detections).
107       When a score above ``activity_threshold`` is detected, switches to
108       full-rate processing for ``active_window_s`` seconds.
109    2. **Silence skipping**: Skip inference when audio RMS is below
110       ``silence_rms`` (no speech possible).
111    3. **Battery-aware**: When on battery and below ``battery_low_pct``,
112       increase the duty cycle factor by ``battery_multiplier``.
113
114    Args:
115        duty_cycle_n: Base duty cycle (process every Nth frame). Default 1 (no skipping).
116        silence_rms: RMS threshold in int16 scale (typical range 0-32768) below which
117            frames are skipped. Default 10.0 filters near-silence.
118        activity_threshold: Score above which the system enters "active" mode. Default 0.3.
119        active_window_s: Seconds to stay in full-rate mode after activity. Default 3.0.
120        battery_low_pct: Battery percent below which power saving kicks in. Default 20.
121        battery_multiplier: Multiply duty_cycle_n by this when on low battery. Default 3.
122        check_battery_interval_s: How often to re-check battery. Default 60.
123    """
124
125    def __init__(
126        self,
127        duty_cycle_n: int = 1,
128        silence_rms: float = 10.0,
129        activity_threshold: float = 0.3,
130        active_window_s: float = 3.0,
131        battery_low_pct: int = 20,
132        battery_multiplier: int = 3,
133        check_battery_interval_s: float = 60.0,
134    ) -> None:
135        if duty_cycle_n < 1:
136            raise ValueError(f"duty_cycle_n must be >= 1, got {duty_cycle_n}")
137
138        self._base_duty = duty_cycle_n
139        self._silence_rms = silence_rms
140        self._activity_threshold = activity_threshold
141        self._active_window_s = active_window_s
142        self._battery_low_pct = battery_low_pct
143        self._battery_multiplier = battery_multiplier
144        self._check_interval = check_battery_interval_s
145
146        # Lock protects all mutable state below
147        self._lock = threading.Lock()
148
149        # State
150        self._frame_counter = 0
151        self._frames_processed = 0
152        self._frames_skipped = 0
153        self._silence_skipped = 0
154        self._last_activity_time = 0.0
155        self._is_active = False
156
157        # Battery state (cached)
158        self._battery_pct = -1
159        self._is_on_battery = False
160        self._last_battery_check = 0.0
161
162    @property
163    def effective_duty_cycle(self) -> int:
164        """Current effective duty cycle considering battery and activity state."""
165        with self._lock:
166            return self._effective_duty_cycle_unlocked()
167
168    def _effective_duty_cycle_unlocked(self) -> int:
169        """Compute duty cycle without acquiring the lock (caller must hold it)."""
170        if self._is_active:
171            return 1  # Full rate when active
172
173        base = self._base_duty
174
175        # Battery scaling
176        if self._is_on_battery and 0 <= self._battery_pct < self._battery_low_pct:
177            base = base * self._battery_multiplier
178
179        return max(1, base)
180
181    def should_process(self, audio_frame: np.ndarray) -> bool:
182        """Decide whether this frame should be processed or skipped.
183
184        Call this before running inference. If it returns False, skip the
185        frame to save CPU/power.
186
187        Args:
188            audio_frame: 1-D audio samples (int16-range float32 or actual int16).
189
190        Returns:
191            True if inference should run on this frame.
192        """
193        # Compute RMS outside the lock (pure computation on immutable input)
194        rms = float(np.sqrt(np.mean(audio_frame.astype(np.float32) ** 2)))
195
196        with self._lock:
197            self._frame_counter += 1
198
199            # Periodically check battery
200            now = time.monotonic()
201            if now - self._last_battery_check > self._check_interval:
202                self._battery_pct, self._is_on_battery = _get_battery_info()
203                self._last_battery_check = now
204
205            # Check if active window has expired
206            if self._is_active and (now - self._last_activity_time > self._active_window_s):
207                self._is_active = False
208
209            # Silence gate: skip if audio is very quiet
210            if rms < self._silence_rms:
211                self._frames_skipped += 1
212                self._silence_skipped += 1
213                return False
214
215            # Duty cycling: process every Nth frame
216            duty = self._effective_duty_cycle_unlocked()
217            if duty > 1 and (self._frame_counter % duty) != 0:
218                self._frames_skipped += 1
219                return False
220
221            self._frames_processed += 1
222            return True
223
224    def report_score(self, score: float) -> None:
225        """Report a detection score to the power manager.
226
227        If the score is above the activity threshold, the manager switches
228        to full-rate processing mode for ``active_window_s`` seconds.
229
230        Args:
231            score: Detection score from the model.
232        """
233        if score >= self._activity_threshold:
234            with self._lock:
235                self._is_active = True
236                self._last_activity_time = time.monotonic()
237
238    def get_state(self) -> PowerState:
239        """Return current power management state snapshot."""
240        with self._lock:
241            total = self._frames_processed + self._frames_skipped
242            # Returns 0.0 if no frames received (no data yet to measure).
243            rate = self._frames_processed / total if total > 0 else 0.0
244
245            return PowerState(
246                battery_percent=self._battery_pct,
247                is_on_battery=self._is_on_battery,
248                duty_cycle_n=self._effective_duty_cycle_unlocked(),
249                frames_processed=self._frames_processed,
250                frames_skipped=self._frames_skipped,
251                silence_skipped=self._silence_skipped,
252                effective_rate=rate,
253            )
254
255    def reset(self) -> None:
256        """Reset all counters and state."""
257        with self._lock:
258            self._frame_counter = 0
259            self._frames_processed = 0
260            self._frames_skipped = 0
261            self._silence_skipped = 0
262            self._last_activity_time = 0.0
263            self._is_active = False

Energy-aware inference controller.

Reduces inference frequency based on battery level, silence detection, and explicit duty cycling configuration.

Modes of power saving:

Duty cycling: Process every Nth frame when idle (no recent detections). When a score above activity_threshold is detected, switches to full-rate processing for active_window_s seconds.
Silence skipping: Skip inference when audio RMS is below silence_rms (no speech possible).
Battery-aware: When on battery and below battery_low_pct, increase the duty cycle factor by battery_multiplier.

Arguments:

duty_cycle_n: Base duty cycle (process every Nth frame). Default 1 (no skipping).
silence_rms: RMS threshold in int16 scale (typical range 0-32768) below which frames are skipped. Default 10.0 filters near-silence.
activity_threshold: Score above which the system enters "active" mode. Default 0.3.
active_window_s: Seconds to stay in full-rate mode after activity. Default 3.0.
battery_low_pct: Battery percent below which power saving kicks in. Default 20.
battery_multiplier: Multiply duty_cycle_n by this when on low battery. Default 3.
check_battery_interval_s: How often to re-check battery. Default 60.

PowerManager( duty_cycle_n: int = 1, silence_rms: float = 10.0, activity_threshold: float = 0.3, active_window_s: float = 3.0, battery_low_pct: int = 20, battery_multiplier: int = 3, check_battery_interval_s: float = 60.0) View Source

125    def __init__(
126        self,
127        duty_cycle_n: int = 1,
128        silence_rms: float = 10.0,
129        activity_threshold: float = 0.3,
130        active_window_s: float = 3.0,
131        battery_low_pct: int = 20,
132        battery_multiplier: int = 3,
133        check_battery_interval_s: float = 60.0,
134    ) -> None:
135        if duty_cycle_n < 1:
136            raise ValueError(f"duty_cycle_n must be >= 1, got {duty_cycle_n}")
137
138        self._base_duty = duty_cycle_n
139        self._silence_rms = silence_rms
140        self._activity_threshold = activity_threshold
141        self._active_window_s = active_window_s
142        self._battery_low_pct = battery_low_pct
143        self._battery_multiplier = battery_multiplier
144        self._check_interval = check_battery_interval_s
145
146        # Lock protects all mutable state below
147        self._lock = threading.Lock()
148
149        # State
150        self._frame_counter = 0
151        self._frames_processed = 0
152        self._frames_skipped = 0
153        self._silence_skipped = 0
154        self._last_activity_time = 0.0
155        self._is_active = False
156
157        # Battery state (cached)
158        self._battery_pct = -1
159        self._is_on_battery = False
160        self._last_battery_check = 0.0

effective_duty_cycle: int View Source

162    @property
163    def effective_duty_cycle(self) -> int:
164        """Current effective duty cycle considering battery and activity state."""
165        with self._lock:
166            return self._effective_duty_cycle_unlocked()

Current effective duty cycle considering battery and activity state.

def should_process(self, audio_frame: numpy.ndarray) -> bool: View Source

181    def should_process(self, audio_frame: np.ndarray) -> bool:
182        """Decide whether this frame should be processed or skipped.
183
184        Call this before running inference. If it returns False, skip the
185        frame to save CPU/power.
186
187        Args:
188            audio_frame: 1-D audio samples (int16-range float32 or actual int16).
189
190        Returns:
191            True if inference should run on this frame.
192        """
193        # Compute RMS outside the lock (pure computation on immutable input)
194        rms = float(np.sqrt(np.mean(audio_frame.astype(np.float32) ** 2)))
195
196        with self._lock:
197            self._frame_counter += 1
198
199            # Periodically check battery
200            now = time.monotonic()
201            if now - self._last_battery_check > self._check_interval:
202                self._battery_pct, self._is_on_battery = _get_battery_info()
203                self._last_battery_check = now
204
205            # Check if active window has expired
206            if self._is_active and (now - self._last_activity_time > self._active_window_s):
207                self._is_active = False
208
209            # Silence gate: skip if audio is very quiet
210            if rms < self._silence_rms:
211                self._frames_skipped += 1
212                self._silence_skipped += 1
213                return False
214
215            # Duty cycling: process every Nth frame
216            duty = self._effective_duty_cycle_unlocked()
217            if duty > 1 and (self._frame_counter % duty) != 0:
218                self._frames_skipped += 1
219                return False
220
221            self._frames_processed += 1
222            return True

Decide whether this frame should be processed or skipped.

Call this before running inference. If it returns False, skip the frame to save CPU/power.

Arguments:

audio_frame: 1-D audio samples (int16-range float32 or actual int16).

Returns:

True if inference should run on this frame.

def report_score(self, score: float) -> None: View Source

224    def report_score(self, score: float) -> None:
225        """Report a detection score to the power manager.
226
227        If the score is above the activity threshold, the manager switches
228        to full-rate processing mode for ``active_window_s`` seconds.
229
230        Args:
231            score: Detection score from the model.
232        """
233        if score >= self._activity_threshold:
234            with self._lock:
235                self._is_active = True
236                self._last_activity_time = time.monotonic()

Report a detection score to the power manager.

If the score is above the activity threshold, the manager switches to full-rate processing mode for active_window_s seconds.

Arguments:

score: Detection score from the model.

def get_state(self) -> violawake_sdk.power_manager.PowerState: View Source

238    def get_state(self) -> PowerState:
239        """Return current power management state snapshot."""
240        with self._lock:
241            total = self._frames_processed + self._frames_skipped
242            # Returns 0.0 if no frames received (no data yet to measure).
243            rate = self._frames_processed / total if total > 0 else 0.0
244
245            return PowerState(
246                battery_percent=self._battery_pct,
247                is_on_battery=self._is_on_battery,
248                duty_cycle_n=self._effective_duty_cycle_unlocked(),
249                frames_processed=self._frames_processed,
250                frames_skipped=self._frames_skipped,
251                silence_skipped=self._silence_skipped,
252                effective_rate=rate,
253            )

Return current power management state snapshot.

def reset(self) -> None: View Source

255    def reset(self) -> None:
256        """Reset all counters and state."""
257        with self._lock:
258            self._frame_counter = 0
259            self._frames_processed = 0
260            self._frames_skipped = 0
261            self._silence_skipped = 0
262            self._last_activity_time = 0.0
263            self._is_active = False

Reset all counters and state.

class VADEngine: View Source

301class VADEngine:
302    """Voice Activity Detection engine.
303
304    Auto-selects the best available backend unless explicitly specified.
305
306    Example::
307
308        vad = VADEngine(backend="silero")  # or "webrtc", "rms", "auto"
309        prob = vad.process_frame(audio_20ms_bytes)
310        is_speech = prob > 0.5
311    """
312
313    def __init__(
314        self,
315        backend: str | VADBackend = VADBackend.AUTO,
316        **backend_kwargs: object,
317    ) -> None:
318        """Initialize the VAD engine.
319
320        Args:
321            backend: One of "auto", "webrtc", "silero", "rms".
322                     "auto" selects the best available backend.
323            **backend_kwargs: Backend-specific arguments.
324                For "silero": no backend-specific args
325                For "webrtc": aggressiveness (0–3, default 2)
326                For "rms": speech_threshold, silence_threshold
327        """
328        if isinstance(backend, str):
329            backend = VADBackend(backend)
330
331        self._backend_name, self._backend = _create_backend(backend, **backend_kwargs)
332
333    @property
334    def backend_name(self) -> str:
335        """Name of the active backend."""
336        return self._backend_name.value
337
338    def process_frame(self, audio: bytes | np.ndarray) -> float:
339        """Process a 16kHz mono audio frame.
340
341        Args:
342            audio: Accepted formats:
343                - bytes/bytearray: int16 PCM
344                - np.ndarray float32/float64: assumed normalized to [-1.0, 1.0],
345                  scaled by 32768 to int16. Use int16 dtype for int16-range data.
346                - np.ndarray int16: converted to bytes directly
347
348            WebRTC accepts 10/20/30ms frames. Silero runs on 512-sample
349            16kHz windows internally and pads/chunks frames as needed.
350
351        Returns:
352            Speech probability in [0.0, 1.0].
353            1.0 = definitely speech, 0.0 = definitely silence.
354        """
355        audio_bytes = _coerce_to_bytes(audio)
356        return self._backend.process_frame(audio_bytes)
357
358    def is_speech(self, audio: bytes | np.ndarray, threshold: float = 0.5) -> bool:
359        """Convenience method: returns True if speech probability exceeds threshold."""
360        return self.process_frame(audio) >= threshold
361
362    def reset(self) -> None:
363        """Reset internal state (useful between utterances)."""
364        self._backend.reset()
365
366    def close(self) -> None:
367        """Release backend resources."""
368        self._backend = None  # type: ignore[assignment]
369
370    def __enter__(self) -> VADEngine:
371        """Enter sync context manager. Returns self."""
372        return self
373
374    def __exit__(
375        self,
376        exc_type: type[BaseException] | None,
377        exc_val: BaseException | None,
378        exc_tb: object,
379    ) -> None:
380        """Exit sync context manager. Releases backend resources."""
381        self.close()

Voice Activity Detection engine.

Auto-selects the best available backend unless explicitly specified.

Example::

vad = VADEngine(backend="silero")  # or "webrtc", "rms", "auto"
prob = vad.process_frame(audio_20ms_bytes)
is_speech = prob > 0.5

VADEngine( backend: str | violawake_sdk.vad.VADBackend = <VADBackend.AUTO: 'auto'>, **backend_kwargs: object) View Source

313    def __init__(
314        self,
315        backend: str | VADBackend = VADBackend.AUTO,
316        **backend_kwargs: object,
317    ) -> None:
318        """Initialize the VAD engine.
319
320        Args:
321            backend: One of "auto", "webrtc", "silero", "rms".
322                     "auto" selects the best available backend.
323            **backend_kwargs: Backend-specific arguments.
324                For "silero": no backend-specific args
325                For "webrtc": aggressiveness (0–3, default 2)
326                For "rms": speech_threshold, silence_threshold
327        """
328        if isinstance(backend, str):
329            backend = VADBackend(backend)
330
331        self._backend_name, self._backend = _create_backend(backend, **backend_kwargs)

Initialize the VAD engine.

Arguments:

backend: One of "auto", "webrtc", "silero", "rms". "auto" selects the best available backend.
**backend_kwargs: Backend-specific arguments. For "silero": no backend-specific args For "webrtc": aggressiveness (0–3, default 2) For "rms": speech_threshold, silence_threshold

backend_name: str View Source

333    @property
334    def backend_name(self) -> str:
335        """Name of the active backend."""
336        return self._backend_name.value

Name of the active backend.

def process_frame(self, audio: bytes | numpy.ndarray) -> float: View Source

338    def process_frame(self, audio: bytes | np.ndarray) -> float:
339        """Process a 16kHz mono audio frame.
340
341        Args:
342            audio: Accepted formats:
343                - bytes/bytearray: int16 PCM
344                - np.ndarray float32/float64: assumed normalized to [-1.0, 1.0],
345                  scaled by 32768 to int16. Use int16 dtype for int16-range data.
346                - np.ndarray int16: converted to bytes directly
347
348            WebRTC accepts 10/20/30ms frames. Silero runs on 512-sample
349            16kHz windows internally and pads/chunks frames as needed.
350
351        Returns:
352            Speech probability in [0.0, 1.0].
353            1.0 = definitely speech, 0.0 = definitely silence.
354        """
355        audio_bytes = _coerce_to_bytes(audio)
356        return self._backend.process_frame(audio_bytes)

Process a 16kHz mono audio frame.

Arguments:

audio: Accepted formats:
- bytes/bytearray: int16 PCM
- np.ndarray float32/float64: assumed normalized to [-1.0, 1.0], scaled by 32768 to int16. Use int16 dtype for int16-range data.
- np.ndarray int16: converted to bytes directly
WebRTC accepts 10/20/30ms frames. Silero runs on 512-sample
16kHz windows internally and pads/chunks frames as needed.

Returns:

Speech probability in [0.0, 1.0]. 1.0 = definitely speech, 0.0 = definitely silence.

def is_speech(self, audio: bytes | numpy.ndarray, threshold: float = 0.5) -> bool: View Source

358    def is_speech(self, audio: bytes | np.ndarray, threshold: float = 0.5) -> bool:
359        """Convenience method: returns True if speech probability exceeds threshold."""
360        return self.process_frame(audio) >= threshold

Convenience method: returns True if speech probability exceeds threshold.

def reset(self) -> None: View Source

362    def reset(self) -> None:
363        """Reset internal state (useful between utterances)."""
364        self._backend.reset()

Reset internal state (useful between utterances).

def close(self) -> None: View Source

366    def close(self) -> None:
367        """Release backend resources."""
368        self._backend = None  # type: ignore[assignment]

Release backend resources.

class TTSEngine: View Source

 61class TTSEngine:
 62    """On-device TTS using Kokoro-82M (Apache 2.0 model).
 63
 64    Thread-safe: multiple threads can call ``synthesize()`` concurrently.
 65    Calls are serialized via ``_synthesis_lock`` since kokoro-onnx is not
 66    guaranteed to be thread-safe. Model initialization is separately guarded
 67    by ``_lock`` (lazy load on first use).
 68
 69    Model files required (auto-downloaded on first use):
 70        - ``kokoro_v1_0.onnx`` — Kokoro-82M model (~326MB)
 71        - ``kokoro_voices_v1_0.bin`` — Voice embeddings (~28MB)
 72
 73    Example::
 74
 75        tts = TTSEngine(voice="af_heart")
 76        audio = tts.synthesize("Hello, world!")  # returns np.ndarray
 77        tts.play(audio)  # blocking by default
 78        tts.play_async(audio)  # optional non-blocking playback
 79    """
 80
 81    def __init__(
 82        self,
 83        voice: str = DEFAULT_VOICE,
 84        speed: float = 1.0,
 85        sample_rate: int = TARGET_SAMPLE_RATE,
 86    ) -> None:
 87        """Initialize the TTS engine.
 88
 89        Args:
 90            voice: Kokoro voice name. Default "af_heart".
 91                   See ``AVAILABLE_VOICES`` for full list.
 92            speed: Speech speed multiplier. 1.0 = normal, 1.2 = 20% faster.
 93            sample_rate: Output sample rate. Default 16kHz (pipeline standard).
 94                         Kokoro outputs 24kHz; resampled if different.
 95        """
 96        if voice not in AVAILABLE_VOICES:
 97            raise ValueError(f"Unknown voice '{voice}'. Available: {', '.join(AVAILABLE_VOICES)}")
 98
 99        if not (0.1 <= speed <= 3.0):
100            raise ValueError(f"Speed must be between 0.1 and 3.0, got {speed}")
101
102        self.voice = voice
103        self.speed = speed
104        self.sample_rate = sample_rate
105        self._lock = threading.Lock()
106        self._synthesis_lock = threading.Lock()
107        self._kokoro: object | None = None
108
109        # Lazy initialization — load model on first use
110        logger.info("TTSEngine created: voice=%s, speed=%.1f", voice, speed)
111
112    def _get_kokoro(self) -> object:
113        """Lazy-load the Kokoro model (thread-safe)."""
114        with self._lock:
115            if self._kokoro is None:
116                self._kokoro = self._load_kokoro()
117        return self._kokoro
118
119    def _load_kokoro(self) -> object:
120        """Load the Kokoro ONNX model."""
121        try:
122            import kokoro_onnx
123        except ImportError as e:
124            raise ImportError(
125                "kokoro-onnx is not installed. Install with: pip install 'violawake[tts]'"
126            ) from e
127
128        try:
129            model_path = get_model_path("kokoro_v1_0")
130            voices_path = get_model_path("kokoro_voices_v1_0")
131        except FileNotFoundError as e:
132            raise ModelNotFoundError(
133                "Kokoro models not found. Run:\n"
134                "  violawake-download --model kokoro_v1_0\n"
135                "  violawake-download --model kokoro_voices_v1_0"
136            ) from e
137
138        try:
139            kokoro = kokoro_onnx.Kokoro(str(model_path), str(voices_path))
140        except Exception as e:
141            raise ModelLoadError(f"Failed to load Kokoro model: {e}") from e
142
143        logger.info("Kokoro-82M loaded: %s", model_path)
144        return kokoro
145
146    def synthesize(self, text: str) -> np.ndarray:
147        """Synthesize text to audio.
148
149        Args:
150            text: Text to synthesize. May be multi-sentence.
151                  Long text is processed as a single batch call.
152
153        Returns:
154            Audio samples as float32 numpy array at ``self.sample_rate``.
155        """
156        if not text.strip():
157            return np.zeros(0, dtype=np.float32)
158
159        kokoro = self._get_kokoro()
160
161        # Hold synthesis lock to serialize access to the kokoro model,
162        # which is not guaranteed to be thread-safe by kokoro-onnx.
163        with self._synthesis_lock:
164            try:
165                # kokoro-onnx API: returns (samples, sample_rate)
166                audio, sr = kokoro.create(  # type: ignore[attr-defined]
167                    text,
168                    voice=self.voice,
169                    speed=self.speed,
170                    lang="en-us",
171                )
172            except Exception as e:
173                logger.exception("TTS synthesis failed for text: %.50s...", text)
174                raise RuntimeError(f"TTS synthesis failed: {e}") from e
175
176        audio = np.asarray(audio, dtype=np.float32)
177
178        # Resample if needed
179        if sr != self.sample_rate:
180            audio = self._resample(audio, sr, self.sample_rate)
181
182        return audio
183
184    def synthesize_chunked(self, text: str) -> Generator[np.ndarray, None, None]:
185        """Synthesize text sentence-by-sentence for lower latency.
186
187        Splits text at sentence boundaries and yields audio for each sentence
188        as soon as it's synthesized. This allows playback to begin before
189        the full text is processed — matching the pattern from production Viola.
190
191        Args:
192            text: Text to synthesize. May be multi-sentence.
193
194        Yields:
195            Audio chunks (one per sentence) as float32 numpy arrays.
196        """
197        sentences = self._split_sentences(text)
198        for sentence in sentences:
199            if sentence.strip():
200                audio = self.synthesize(sentence)
201                if audio.size > 0:
202                    yield audio
203
204    def play(self, audio: np.ndarray, *, blocking: bool = True) -> None:
205        """Play audio through the default output device.
206
207        Args:
208            audio: Float32 numpy array of audio samples.
209            blocking: If True, wait for playback to finish. If False, return
210                      immediately after starting playback.
211        """
212        try:
213            import sounddevice as sd
214        except ImportError as sd_err:
215            logger.debug("sounddevice not available (%s), falling back to pyaudio", sd_err)
216            try:
217                self._play_pyaudio(audio, blocking=blocking)
218            except ImportError as e:
219                raise ImportError(
220                    "No audio playback backend is installed. "
221                    "Install sounddevice with: pip install sounddevice "
222                    "or install violawake[audio] for PyAudio playback."
223                ) from e
224            return
225
226        # Copy to prevent mutation of caller's array during async playback
227        sd.play(audio.copy(), samplerate=self.sample_rate, blocking=blocking)
228
229    def play_async(self, audio: np.ndarray) -> None:
230        """Play audio without blocking the calling thread."""
231        self.play(audio, blocking=False)
232
233    def _play_pyaudio(self, audio: np.ndarray, *, blocking: bool = True) -> None:
234        """Play audio using pyaudio as fallback."""
235        try:
236            import pyaudio
237        except ImportError:
238            raise ImportError(
239                "pyaudio is required for audio playback. Install with: pip install violawake[audio]"
240            ) from None
241
242        if not blocking:
243            thread = threading.Thread(
244                target=self._play_pyaudio,
245                args=(audio.copy(),),
246                kwargs={"blocking": True},
247                daemon=True,
248            )
249            thread.start()
250            return
251
252        clipped = np.clip(audio, -1.0, 1.0)
253        pcm = (clipped * 32767).astype(np.int16)
254        pa = pyaudio.PyAudio()
255        stream = pa.open(
256            format=pyaudio.paInt16,
257            channels=1,
258            rate=self.sample_rate,
259            output=True,
260        )
261        try:
262            stream.write(pcm.tobytes())
263        finally:
264            stream.stop_stream()
265            stream.close()
266            pa.terminate()
267
268    @staticmethod
269    def _resample(audio: np.ndarray, src_rate: int, dst_rate: int) -> np.ndarray:
270        """Resample audio using scipy."""
271        import math
272
273        try:
274            from scipy.signal import resample_poly
275        except ImportError as e:
276            raise ImportError(
277                "scipy is required for audio resampling. Install with: pip install scipy"
278            ) from e
279        gcd = math.gcd(src_rate, dst_rate)
280        return resample_poly(audio, dst_rate // gcd, src_rate // gcd).astype(np.float32)
281
282    @staticmethod
283    def _split_sentences(text: str) -> list[str]:
284        """Split text at sentence boundaries for chunked synthesis.
285
286        Uses ``pysbd`` when available for robust sentence boundary
287        disambiguation, with a regex fallback if the dependency is missing.
288        """
289        if not text:
290            return []
291
292        if pysbd is not None:
293            segmenter = pysbd.Segmenter(language="en", clean=False)
294            return [s.strip() for s in segmenter.segment(text) if s and s.strip()]
295
296        return TTSEngine._split_sentences_fallback(text)
297
298    @staticmethod
299    def _split_sentences_fallback(text: str) -> list[str]:
300        """Fallback regex sentence splitter used when ``pysbd`` is unavailable."""
301        import re
302
303        pattern = r"(?<=[.!?])\s+(?=[A-Z])|(?<=[.!?])\s*$"
304        parts = re.split(pattern, text)
305        return [s.strip() for s in parts if s and s.strip()]
306
307    def close(self) -> None:
308        """Release model resources."""
309        self._kokoro = None
310
311    def __enter__(self) -> TTSEngine:
312        """Enter sync context manager. Returns self."""
313        return self
314
315    def __exit__(
316        self,
317        exc_type: type[BaseException] | None,
318        exc_val: BaseException | None,
319        exc_tb: object,
320    ) -> None:
321        """Exit sync context manager. Releases model resources."""
322        self.close()

On-device TTS using Kokoro-82M (Apache 2.0 model).

Thread-safe: multiple threads can call synthesize() concurrently. Calls are serialized via _synthesis_lock since kokoro-onnx is not guaranteed to be thread-safe. Model initialization is separately guarded by _lock (lazy load on first use).

Model files required (auto-downloaded on first use): - kokoro_v1_0.onnx — Kokoro-82M model (~326MB) - kokoro_voices_v1_0.bin — Voice embeddings (~28MB)

Example::

tts = TTSEngine(voice="af_heart")
audio = tts.synthesize("Hello, world!")  # returns np.ndarray
tts.play(audio)  # blocking by default
tts.play_async(audio)  # optional non-blocking playback

TTSEngine( voice: str = 'af_heart', speed: float = 1.0, sample_rate: int = 16000) View Source

 81    def __init__(
 82        self,
 83        voice: str = DEFAULT_VOICE,
 84        speed: float = 1.0,
 85        sample_rate: int = TARGET_SAMPLE_RATE,
 86    ) -> None:
 87        """Initialize the TTS engine.
 88
 89        Args:
 90            voice: Kokoro voice name. Default "af_heart".
 91                   See ``AVAILABLE_VOICES`` for full list.
 92            speed: Speech speed multiplier. 1.0 = normal, 1.2 = 20% faster.
 93            sample_rate: Output sample rate. Default 16kHz (pipeline standard).
 94                         Kokoro outputs 24kHz; resampled if different.
 95        """
 96        if voice not in AVAILABLE_VOICES:
 97            raise ValueError(f"Unknown voice '{voice}'. Available: {', '.join(AVAILABLE_VOICES)}")
 98
 99        if not (0.1 <= speed <= 3.0):
100            raise ValueError(f"Speed must be between 0.1 and 3.0, got {speed}")
101
102        self.voice = voice
103        self.speed = speed
104        self.sample_rate = sample_rate
105        self._lock = threading.Lock()
106        self._synthesis_lock = threading.Lock()
107        self._kokoro: object | None = None
108
109        # Lazy initialization — load model on first use
110        logger.info("TTSEngine created: voice=%s, speed=%.1f", voice, speed)

Initialize the TTS engine.

Arguments:

voice: Kokoro voice name. Default "af_heart". See AVAILABLE_VOICES for full list.
speed: Speech speed multiplier. 1.0 = normal, 1.2 = 20% faster.
sample_rate: Output sample rate. Default 16kHz (pipeline standard). Kokoro outputs 24kHz; resampled if different.

voice

speed

sample_rate

def synthesize(self, text: str) -> numpy.ndarray: View Source

146    def synthesize(self, text: str) -> np.ndarray:
147        """Synthesize text to audio.
148
149        Args:
150            text: Text to synthesize. May be multi-sentence.
151                  Long text is processed as a single batch call.
152
153        Returns:
154            Audio samples as float32 numpy array at ``self.sample_rate``.
155        """
156        if not text.strip():
157            return np.zeros(0, dtype=np.float32)
158
159        kokoro = self._get_kokoro()
160
161        # Hold synthesis lock to serialize access to the kokoro model,
162        # which is not guaranteed to be thread-safe by kokoro-onnx.
163        with self._synthesis_lock:
164            try:
165                # kokoro-onnx API: returns (samples, sample_rate)
166                audio, sr = kokoro.create(  # type: ignore[attr-defined]
167                    text,
168                    voice=self.voice,
169                    speed=self.speed,
170                    lang="en-us",
171                )
172            except Exception as e:
173                logger.exception("TTS synthesis failed for text: %.50s...", text)
174                raise RuntimeError(f"TTS synthesis failed: {e}") from e
175
176        audio = np.asarray(audio, dtype=np.float32)
177
178        # Resample if needed
179        if sr != self.sample_rate:
180            audio = self._resample(audio, sr, self.sample_rate)
181
182        return audio

Synthesize text to audio.

Arguments:

text: Text to synthesize. May be multi-sentence. Long text is processed as a single batch call.

Returns:

Audio samples as float32 numpy array at self.sample_rate.

def synthesize_chunked(self, text: str) -> Generator[numpy.ndarray, None, None]: View Source

184    def synthesize_chunked(self, text: str) -> Generator[np.ndarray, None, None]:
185        """Synthesize text sentence-by-sentence for lower latency.
186
187        Splits text at sentence boundaries and yields audio for each sentence
188        as soon as it's synthesized. This allows playback to begin before
189        the full text is processed — matching the pattern from production Viola.
190
191        Args:
192            text: Text to synthesize. May be multi-sentence.
193
194        Yields:
195            Audio chunks (one per sentence) as float32 numpy arrays.
196        """
197        sentences = self._split_sentences(text)
198        for sentence in sentences:
199            if sentence.strip():
200                audio = self.synthesize(sentence)
201                if audio.size > 0:
202                    yield audio

Synthesize text sentence-by-sentence for lower latency.

Splits text at sentence boundaries and yields audio for each sentence as soon as it's synthesized. This allows playback to begin before the full text is processed — matching the pattern from production Viola.

Arguments:

text: Text to synthesize. May be multi-sentence.

Yields:

Audio chunks (one per sentence) as float32 numpy arrays.

def play(self, audio: numpy.ndarray, *, blocking: bool = True) -> None: View Source

204    def play(self, audio: np.ndarray, *, blocking: bool = True) -> None:
205        """Play audio through the default output device.
206
207        Args:
208            audio: Float32 numpy array of audio samples.
209            blocking: If True, wait for playback to finish. If False, return
210                      immediately after starting playback.
211        """
212        try:
213            import sounddevice as sd
214        except ImportError as sd_err:
215            logger.debug("sounddevice not available (%s), falling back to pyaudio", sd_err)
216            try:
217                self._play_pyaudio(audio, blocking=blocking)
218            except ImportError as e:
219                raise ImportError(
220                    "No audio playback backend is installed. "
221                    "Install sounddevice with: pip install sounddevice "
222                    "or install violawake[audio] for PyAudio playback."
223                ) from e
224            return
225
226        # Copy to prevent mutation of caller's array during async playback
227        sd.play(audio.copy(), samplerate=self.sample_rate, blocking=blocking)

Play audio through the default output device.

Arguments:

audio: Float32 numpy array of audio samples.
blocking: If True, wait for playback to finish. If False, return immediately after starting playback.

def play_async(self, audio: numpy.ndarray) -> None: View Source

229    def play_async(self, audio: np.ndarray) -> None:
230        """Play audio without blocking the calling thread."""
231        self.play(audio, blocking=False)

Play audio without blocking the calling thread.

def close(self) -> None: View Source

307    def close(self) -> None:
308        """Release model resources."""
309        self._kokoro = None

Release model resources.

class STTEngine: View Source

 85class STTEngine:
 86    """Speech-to-text transcription via faster-whisper.
 87
 88    Thread-safe: ``WhisperModel`` is thread-safe for concurrent ``transcribe()`` calls.
 89
 90    Model is loaded once and reused. First call includes model load time
 91    (~1-3s). Subsequent calls are ~380ms (base model, CPU, 3s audio).
 92
 93    Example::
 94
 95        stt = STTEngine(model="base")
 96        text = stt.transcribe(audio_np_float32)
 97        print(text)  # "what's the weather today"
 98    """
 99
100    def __init__(
101        self,
102        model: str = DEFAULT_MODEL,
103        device: str = "cpu",
104        compute_type: str = "int8",
105        language: str | None = None,
106        language_cache_ttl_s: float = 60.0,
107    ) -> None:
108        """Initialize the STT engine.
109
110        Args:
111            model: Whisper model size. One of: tiny, base, small, medium, large-v3.
112                   Default "base" — good accuracy/speed balance (WER ~9%).
113            device: "cpu" or "cuda". Default "cpu".
114            compute_type: CTranslate2 compute type. "int8" (default), "float16", "float32".
115                          "int8" is fastest on CPU with minimal accuracy loss.
116            language: Force a specific language (e.g., "en"). None = auto-detect.
117            language_cache_ttl_s: Cache detected language for N seconds to avoid
118                                   per-call language detection overhead.
119        """
120        if model not in MODEL_PROFILES:
121            available = ", ".join(MODEL_PROFILES.keys())
122            raise ValueError(f"Unknown model '{model}'. Available: {available}")
123
124        self.model_name = model
125        self.device = device
126        self.compute_type = compute_type
127        self.forced_language = language
128        self._language_cache: tuple[str, float] | None = None  # (lang, cached_at)
129        self._language_cache_ttl = language_cache_ttl_s
130        self._model: WhisperModel | None = None
131        self._model_lock = threading.Lock()
132
133        profile = MODEL_PROFILES[model]
134        logger.info(
135            "STTEngine created: model=%s, device=%s (WER~%.0f%%, %dMB)",
136            model,
137            device,
138            profile["wer"],
139            profile["vram_mb"],
140        )
141
142    def _get_model(self) -> WhisperModel:
143        """Lazy-load the Whisper model on first use (thread-safe)."""
144        if self._model is not None:
145            return self._model
146
147        with self._model_lock:
148            # Double-checked locking: another thread may have loaded
149            # the model while we waited for the lock.
150            if self._model is not None:
151                return self._model
152
153            try:
154                from faster_whisper import WhisperModel  # type: ignore[import]
155            except ModuleNotFoundError as e:
156                if e.name == "faster_whisper":
157                    raise ImportError(
158                        "faster-whisper is not installed. "
159                        "Install with: pip install 'violawake[stt]'"
160                    ) from e
161                raise ImportError(
162                    f"faster-whisper is installed but failed to import dependency '{e.name}': {e}"
163                ) from e
164            except ImportError as e:
165                raise ImportError(f"faster-whisper is installed but failed to import: {e}") from e
166
167            logger.info("Loading Whisper model '%s'...", self.model_name)
168            t0 = time.perf_counter()
169            self._model = WhisperModel(
170                self.model_name,
171                device=self.device,
172                compute_type=self.compute_type,
173            )
174            elapsed_ms = (time.perf_counter() - t0) * 1000
175            logger.info("Whisper model loaded in %.0f ms", elapsed_ms)
176
177        return self._model
178
179    def transcribe(self, audio: np.ndarray) -> str:
180        """Transcribe audio to text.
181
182        Note:
183            This engine uses a progressive temperature fallback of
184            ``(0.0, 0.2, 0.4)`` during decoding, which can trigger up to
185            3 decoding passes and increase latency. For
186            low-latency use cases, prefer a single-pass configuration such as
187            ``temperature_fallback=[0.0]``.
188
189        Args:
190            audio: Float32 numpy array at 16kHz mono. Values should be in [-1.0, 1.0].
191
192        Returns:
193            Transcribed text as string. Empty string if no speech detected.
194        """
195        result = self.transcribe_full(audio)
196        return result.text
197
198    def transcribe_streaming(
199        self,
200        audio: np.ndarray,
201        channels_first: bool | None = None,
202        beam_size: int = 5,
203        best_of: int = 5,
204        temperature: list[float] | None = None,
205    ) -> Iterator[TranscriptSegment]:
206        """Stream transcription segments as they become available.
207
208        Uses faster-whisper's generator mode: ``model.transcribe()`` returns a
209        ``(segments_iterator, info)`` tuple.  This method yields each
210        ``TranscriptSegment`` one at a time as faster-whisper decodes it,
211        instead of collecting all segments first.
212
213        This is useful when:
214        - You want to display partial results before full transcription completes.
215        - You need to pipe segments to a downstream consumer (TTS, logging, etc.)
216          without waiting for the full buffer to finish.
217
218        Note:
219            Segments with ``no_speech_prob`` above ``NO_SPEECH_THRESHOLD`` are
220            silently skipped (not yielded).
221
222        Args:
223            audio: Float32 numpy array at 16kHz mono, or 2-D stereo.
224            channels_first: Layout hint for 2-D stereo audio (same semantics as
225                ``transcribe_full``).
226            beam_size: Beam search width. Default 5.
227            best_of: Number of candidates when sampling. Default 5.
228            temperature: Temperature schedule. Default ``(0.0, 0.2, 0.4)``.
229
230        Yields:
231            TranscriptSegment — one per decoded segment, in time order.
232
233        Example::
234
235            stt = STTEngine(model="base")
236            for seg in stt.transcribe_streaming(audio_np):
237                print(f"[{seg.start:.1f}s] {seg.text}")
238        """
239        if temperature is None:
240            # Keep 0.0 inside a fallback schedule so faster-whisper can retry if needed
241            # instead of pairing greedy decoding with beam_size > 1 in a single pass.
242            temperature = list(_DEFAULT_TEMPERATURE_FALLBACK)
243
244        audio = np.asarray(audio, dtype=np.float32)
245        if audio.ndim > 1:
246            if channels_first is True:
247                audio = audio.mean(axis=0)
248            elif channels_first is False:
249                audio = audio.mean(axis=1)
250            else:
251                if audio.shape[0] < audio.shape[1]:
252                    audio = audio.mean(axis=0)
253                else:
254                    audio = audio.mean(axis=1)
255
256        language = self._get_language()
257        model = self._get_model()
258
259        logger.debug("transcribe_streaming: starting generator on %d samples", len(audio))
260
261        segments_gen, info = model.transcribe(
262            audio,
263            language=language,
264            vad_filter=True,
265            vad_parameters={"min_silence_duration_ms": 500},
266            word_timestamps=False,
267            beam_size=beam_size,
268            best_of=best_of,
269            temperature=temperature,
270        )
271
272        # Update language cache after model.transcribe() returns info — same
273        # logic as transcribe_full, but we must do it before consuming the
274        # generator so the cache is primed for subsequent calls.
275        if language is None and info.language_probability > 0.5:
276            with self._model_lock:
277                self._language_cache = (info.language, time.monotonic())
278
279        for seg in segments_gen:
280            if seg.no_speech_prob > NO_SPEECH_THRESHOLD:
281                logger.debug(
282                    "Skipping silent segment [%.1f-%.1f] no_speech_prob=%.2f",
283                    seg.start,
284                    seg.end,
285                    seg.no_speech_prob,
286                )
287                continue
288
289            text = seg.text.strip()
290            logger.debug("Streaming segment [%.1f-%.1f]: '%s'", seg.start, seg.end, text)
291            yield TranscriptSegment(
292                text=text,
293                start=seg.start,
294                end=seg.end,
295                no_speech_prob=seg.no_speech_prob,
296            )
297
298    def transcribe_full(
299        self,
300        audio: np.ndarray,
301        channels_first: bool | None = None,
302    ) -> TranscriptResult:
303        """Transcribe audio and return full result with segments, timing, and metadata.
304
305        Args:
306            audio: Float32 numpy array at 16kHz mono, or 2-D stereo.
307            channels_first: Layout hint for 2-D stereo audio.
308                ``True``  = (channels, samples)  e.g. shape (2, 48000).
309                ``False`` = (samples, channels)  e.g. shape (48000, 2) — the
310                standard layout.
311                ``None`` (default) = fall back to a shape heuristic (smaller
312                dimension is assumed to be channels).  Prefer passing an
313                explicit value to avoid ambiguity with short audio clips.
314
315        Returns:
316            TranscriptResult with text, segments, language, and no_speech_prob.
317        """
318        audio = np.asarray(audio, dtype=np.float32)
319        if audio.ndim > 1:
320            if channels_first is True:
321                # Explicit: (channels, samples) — e.g. shape (2, 48000)
322                audio = audio.mean(axis=0)
323            elif channels_first is False:
324                # Explicit: (samples, channels) — e.g. shape (48000, 2)
325                audio = audio.mean(axis=1)
326            else:
327                # Legacy heuristic: channels axis is the smaller dimension.
328                if audio.shape[0] < audio.shape[1]:
329                    audio = audio.mean(axis=0)
330                else:
331                    audio = audio.mean(axis=1)
332
333        # Determine language (use cache if available)
334        language = self._get_language()
335
336        model = self._get_model()
337        t0 = time.perf_counter()
338
339        segments_gen, info = model.transcribe(
340            audio,
341            language=language,
342            vad_filter=True,  # Use Silero VAD for silence removal
343            vad_parameters={"min_silence_duration_ms": 500},
344            word_timestamps=False,
345            beam_size=5,
346            best_of=5,
347            # Keep 0.0 in a short fallback schedule so the initial pass is greedy-safe
348            # without combining a lone temperature=0.0 decode with beam_size > 1.
349            temperature=_DEFAULT_TEMPERATURE_FALLBACK,
350        )
351
352        # Consume the generator (transcription happens here)
353        segments = list(segments_gen)
354        elapsed_ms = (time.perf_counter() - t0) * 1000
355
356        # Update language cache (protected by _model_lock for thread safety)
357        if language is None and info.language_probability > 0.5:
358            with self._model_lock:
359                self._language_cache = (info.language, time.monotonic())
360
361        transcript_segments = [
362            TranscriptSegment(
363                text=s.text.strip(),
364                start=s.start,
365                end=s.end,
366                no_speech_prob=s.no_speech_prob,
367            )
368            for s in segments
369        ]
370
371        full_text = " ".join(s.text for s in transcript_segments).strip()
372        overall_no_speech = max((s.no_speech_prob for s in transcript_segments), default=0.0)
373
374        if overall_no_speech > NO_SPEECH_THRESHOLD:
375            logger.debug(
376                "No speech detected (no_speech_prob=%.2f) — returning empty",
377                overall_no_speech,
378            )
379            full_text = ""
380
381        logger.debug(
382            "Transcribed in %.0f ms: '%s'",
383            elapsed_ms,
384            full_text[:60] + "..." if len(full_text) > 60 else full_text,
385        )
386
387        return TranscriptResult(
388            text=full_text,
389            segments=transcript_segments,
390            language=info.language,
391            language_prob=info.language_probability,
392            duration_s=info.duration,
393            no_speech_prob=overall_no_speech,
394        )
395
396    def _get_language(self) -> str | None:
397        """Return cached language or None for auto-detection.
398
399        Thread-safe: reads ``_language_cache`` under ``_model_lock``.
400        """
401        if self.forced_language:
402            return self.forced_language
403
404        with self._model_lock:
405            if self._language_cache is not None:
406                lang, cached_at = self._language_cache
407                if time.monotonic() - cached_at < self._language_cache_ttl:
408                    return lang
409
410        return None  # auto-detect
411
412    def prewarm(self) -> None:
413        """Load the model eagerly (avoids cold-start latency on first transcription)."""
414        self._get_model()
415        logger.info("STTEngine prewarmed: model '%s' loaded", self.model_name)
416
417    def close(self) -> None:
418        """Release model resources."""
419        with self._model_lock:
420            self._model = None
421
422    def __enter__(self) -> STTEngine:
423        """Enter sync context manager. Returns self."""
424        return self
425
426    def __exit__(
427        self,
428        exc_type: type[BaseException] | None,
429        exc_val: BaseException | None,
430        exc_tb: object,
431    ) -> None:
432        """Exit sync context manager. Releases model resources."""
433        self.close()

Speech-to-text transcription via faster-whisper.

Thread-safe: WhisperModel is thread-safe for concurrent transcribe() calls.

Model is loaded once and reused. First call includes model load time (~1-3s). Subsequent calls are ~380ms (base model, CPU, 3s audio).

Example::

stt = STTEngine(model="base")
text = stt.transcribe(audio_np_float32)
print(text)  # "what's the weather today"

STTEngine( model: str = 'base', device: str = 'cpu', compute_type: str = 'int8', language: str | None = None, language_cache_ttl_s: float = 60.0) View Source

100    def __init__(
101        self,
102        model: str = DEFAULT_MODEL,
103        device: str = "cpu",
104        compute_type: str = "int8",
105        language: str | None = None,
106        language_cache_ttl_s: float = 60.0,
107    ) -> None:
108        """Initialize the STT engine.
109
110        Args:
111            model: Whisper model size. One of: tiny, base, small, medium, large-v3.
112                   Default "base" — good accuracy/speed balance (WER ~9%).
113            device: "cpu" or "cuda". Default "cpu".
114            compute_type: CTranslate2 compute type. "int8" (default), "float16", "float32".
115                          "int8" is fastest on CPU with minimal accuracy loss.
116            language: Force a specific language (e.g., "en"). None = auto-detect.
117            language_cache_ttl_s: Cache detected language for N seconds to avoid
118                                   per-call language detection overhead.
119        """
120        if model not in MODEL_PROFILES:
121            available = ", ".join(MODEL_PROFILES.keys())
122            raise ValueError(f"Unknown model '{model}'. Available: {available}")
123
124        self.model_name = model
125        self.device = device
126        self.compute_type = compute_type
127        self.forced_language = language
128        self._language_cache: tuple[str, float] | None = None  # (lang, cached_at)
129        self._language_cache_ttl = language_cache_ttl_s
130        self._model: WhisperModel | None = None
131        self._model_lock = threading.Lock()
132
133        profile = MODEL_PROFILES[model]
134        logger.info(
135            "STTEngine created: model=%s, device=%s (WER~%.0f%%, %dMB)",
136            model,
137            device,
138            profile["wer"],
139            profile["vram_mb"],
140        )

Initialize the STT engine.

Arguments:

model: Whisper model size. One of: tiny, base, small, medium, large-v3. Default "base" — good accuracy/speed balance (WER ~9%).
device: "cpu" or "cuda". Default "cpu".
compute_type: CTranslate2 compute type. "int8" (default), "float16", "float32". "int8" is fastest on CPU with minimal accuracy loss.
language: Force a specific language (e.g., "en"). None = auto-detect.
language_cache_ttl_s: Cache detected language for N seconds to avoid per-call language detection overhead.

model_name

device

compute_type

forced_language

def transcribe(self, audio: numpy.ndarray) -> str: View Source

179    def transcribe(self, audio: np.ndarray) -> str:
180        """Transcribe audio to text.
181
182        Note:
183            This engine uses a progressive temperature fallback of
184            ``(0.0, 0.2, 0.4)`` during decoding, which can trigger up to
185            3 decoding passes and increase latency. For
186            low-latency use cases, prefer a single-pass configuration such as
187            ``temperature_fallback=[0.0]``.
188
189        Args:
190            audio: Float32 numpy array at 16kHz mono. Values should be in [-1.0, 1.0].
191
192        Returns:
193            Transcribed text as string. Empty string if no speech detected.
194        """
195        result = self.transcribe_full(audio)
196        return result.text

Transcribe audio to text.

Note:

This engine uses a progressive temperature fallback of (0.0, 0.2, 0.4) during decoding, which can trigger up to 3 decoding passes and increase latency. For low-latency use cases, prefer a single-pass configuration such as temperature_fallback=[0.0].

Arguments:

audio: Float32 numpy array at 16kHz mono. Values should be in [-1.0, 1.0].

Returns:

Transcribed text as string. Empty string if no speech detected.

def transcribe_streaming( self, audio: numpy.ndarray, channels_first: bool | None = None, beam_size: int = 5, best_of: int = 5, temperature: list[float] | None = None) -> Iterator[violawake_sdk.stt.TranscriptSegment]: View Source

198    def transcribe_streaming(
199        self,
200        audio: np.ndarray,
201        channels_first: bool | None = None,
202        beam_size: int = 5,
203        best_of: int = 5,
204        temperature: list[float] | None = None,
205    ) -> Iterator[TranscriptSegment]:
206        """Stream transcription segments as they become available.
207
208        Uses faster-whisper's generator mode: ``model.transcribe()`` returns a
209        ``(segments_iterator, info)`` tuple.  This method yields each
210        ``TranscriptSegment`` one at a time as faster-whisper decodes it,
211        instead of collecting all segments first.
212
213        This is useful when:
214        - You want to display partial results before full transcription completes.
215        - You need to pipe segments to a downstream consumer (TTS, logging, etc.)
216          without waiting for the full buffer to finish.
217
218        Note:
219            Segments with ``no_speech_prob`` above ``NO_SPEECH_THRESHOLD`` are
220            silently skipped (not yielded).
221
222        Args:
223            audio: Float32 numpy array at 16kHz mono, or 2-D stereo.
224            channels_first: Layout hint for 2-D stereo audio (same semantics as
225                ``transcribe_full``).
226            beam_size: Beam search width. Default 5.
227            best_of: Number of candidates when sampling. Default 5.
228            temperature: Temperature schedule. Default ``(0.0, 0.2, 0.4)``.
229
230        Yields:
231            TranscriptSegment — one per decoded segment, in time order.
232
233        Example::
234
235            stt = STTEngine(model="base")
236            for seg in stt.transcribe_streaming(audio_np):
237                print(f"[{seg.start:.1f}s] {seg.text}")
238        """
239        if temperature is None:
240            # Keep 0.0 inside a fallback schedule so faster-whisper can retry if needed
241            # instead of pairing greedy decoding with beam_size > 1 in a single pass.
242            temperature = list(_DEFAULT_TEMPERATURE_FALLBACK)
243
244        audio = np.asarray(audio, dtype=np.float32)
245        if audio.ndim > 1:
246            if channels_first is True:
247                audio = audio.mean(axis=0)
248            elif channels_first is False:
249                audio = audio.mean(axis=1)
250            else:
251                if audio.shape[0] < audio.shape[1]:
252                    audio = audio.mean(axis=0)
253                else:
254                    audio = audio.mean(axis=1)
255
256        language = self._get_language()
257        model = self._get_model()
258
259        logger.debug("transcribe_streaming: starting generator on %d samples", len(audio))
260
261        segments_gen, info = model.transcribe(
262            audio,
263            language=language,
264            vad_filter=True,
265            vad_parameters={"min_silence_duration_ms": 500},
266            word_timestamps=False,
267            beam_size=beam_size,
268            best_of=best_of,
269            temperature=temperature,
270        )
271
272        # Update language cache after model.transcribe() returns info — same
273        # logic as transcribe_full, but we must do it before consuming the
274        # generator so the cache is primed for subsequent calls.
275        if language is None and info.language_probability > 0.5:
276            with self._model_lock:
277                self._language_cache = (info.language, time.monotonic())
278
279        for seg in segments_gen:
280            if seg.no_speech_prob > NO_SPEECH_THRESHOLD:
281                logger.debug(
282                    "Skipping silent segment [%.1f-%.1f] no_speech_prob=%.2f",
283                    seg.start,
284                    seg.end,
285                    seg.no_speech_prob,
286                )
287                continue
288
289            text = seg.text.strip()
290            logger.debug("Streaming segment [%.1f-%.1f]: '%s'", seg.start, seg.end, text)
291            yield TranscriptSegment(
292                text=text,
293                start=seg.start,
294                end=seg.end,
295                no_speech_prob=seg.no_speech_prob,
296            )

Stream transcription segments as they become available.

Uses faster-whisper's generator mode: model.transcribe() returns a (segments_iterator, info) tuple. This method yields each TranscriptSegment one at a time as faster-whisper decodes it, instead of collecting all segments first.

This is useful when:

You want to display partial results before full transcription completes.
You need to pipe segments to a downstream consumer (TTS, logging, etc.) without waiting for the full buffer to finish.

Note:

Segments with no_speech_prob above NO_SPEECH_THRESHOLD are silently skipped (not yielded).

Arguments:

audio: Float32 numpy array at 16kHz mono, or 2-D stereo.
channels_first: Layout hint for 2-D stereo audio (same semantics as transcribe_full).
beam_size: Beam search width. Default 5.
best_of: Number of candidates when sampling. Default 5.
temperature: Temperature schedule. Default (0.0, 0.2, 0.4).

Yields:

TranscriptSegment — one per decoded segment, in time order.

Example::

stt = STTEngine(model="base")
for seg in stt.transcribe_streaming(audio_np):
    print(f"[{seg.start:.1f}s] {seg.text}")

def transcribe_full( self, audio: numpy.ndarray, channels_first: bool | None = None) -> violawake_sdk.stt.TranscriptResult: View Source

298    def transcribe_full(
299        self,
300        audio: np.ndarray,
301        channels_first: bool | None = None,
302    ) -> TranscriptResult:
303        """Transcribe audio and return full result with segments, timing, and metadata.
304
305        Args:
306            audio: Float32 numpy array at 16kHz mono, or 2-D stereo.
307            channels_first: Layout hint for 2-D stereo audio.
308                ``True``  = (channels, samples)  e.g. shape (2, 48000).
309                ``False`` = (samples, channels)  e.g. shape (48000, 2) — the
310                standard layout.
311                ``None`` (default) = fall back to a shape heuristic (smaller
312                dimension is assumed to be channels).  Prefer passing an
313                explicit value to avoid ambiguity with short audio clips.
314
315        Returns:
316            TranscriptResult with text, segments, language, and no_speech_prob.
317        """
318        audio = np.asarray(audio, dtype=np.float32)
319        if audio.ndim > 1:
320            if channels_first is True:
321                # Explicit: (channels, samples) — e.g. shape (2, 48000)
322                audio = audio.mean(axis=0)
323            elif channels_first is False:
324                # Explicit: (samples, channels) — e.g. shape (48000, 2)
325                audio = audio.mean(axis=1)
326            else:
327                # Legacy heuristic: channels axis is the smaller dimension.
328                if audio.shape[0] < audio.shape[1]:
329                    audio = audio.mean(axis=0)
330                else:
331                    audio = audio.mean(axis=1)
332
333        # Determine language (use cache if available)
334        language = self._get_language()
335
336        model = self._get_model()
337        t0 = time.perf_counter()
338
339        segments_gen, info = model.transcribe(
340            audio,
341            language=language,
342            vad_filter=True,  # Use Silero VAD for silence removal
343            vad_parameters={"min_silence_duration_ms": 500},
344            word_timestamps=False,
345            beam_size=5,
346            best_of=5,
347            # Keep 0.0 in a short fallback schedule so the initial pass is greedy-safe
348            # without combining a lone temperature=0.0 decode with beam_size > 1.
349            temperature=_DEFAULT_TEMPERATURE_FALLBACK,
350        )
351
352        # Consume the generator (transcription happens here)
353        segments = list(segments_gen)
354        elapsed_ms = (time.perf_counter() - t0) * 1000
355
356        # Update language cache (protected by _model_lock for thread safety)
357        if language is None and info.language_probability > 0.5:
358            with self._model_lock:
359                self._language_cache = (info.language, time.monotonic())
360
361        transcript_segments = [
362            TranscriptSegment(
363                text=s.text.strip(),
364                start=s.start,
365                end=s.end,
366                no_speech_prob=s.no_speech_prob,
367            )
368            for s in segments
369        ]
370
371        full_text = " ".join(s.text for s in transcript_segments).strip()
372        overall_no_speech = max((s.no_speech_prob for s in transcript_segments), default=0.0)
373
374        if overall_no_speech > NO_SPEECH_THRESHOLD:
375            logger.debug(
376                "No speech detected (no_speech_prob=%.2f) — returning empty",
377                overall_no_speech,
378            )
379            full_text = ""
380
381        logger.debug(
382            "Transcribed in %.0f ms: '%s'",
383            elapsed_ms,
384            full_text[:60] + "..." if len(full_text) > 60 else full_text,
385        )
386
387        return TranscriptResult(
388            text=full_text,
389            segments=transcript_segments,
390            language=info.language,
391            language_prob=info.language_probability,
392            duration_s=info.duration,
393            no_speech_prob=overall_no_speech,
394        )

Transcribe audio and return full result with segments, timing, and metadata.

Arguments:

audio: Float32 numpy array at 16kHz mono, or 2-D stereo.
channels_first: Layout hint for 2-D stereo audio. True = (channels, samples) e.g. shape (2, 48000). False = (samples, channels) e.g. shape (48000, 2) — the standard layout. None (default) = fall back to a shape heuristic (smaller dimension is assumed to be channels). Prefer passing an explicit value to avoid ambiguity with short audio clips.

Returns:

TranscriptResult with text, segments, language, and no_speech_prob.

def prewarm(self) -> None: View Source

412    def prewarm(self) -> None:
413        """Load the model eagerly (avoids cold-start latency on first transcription)."""
414        self._get_model()
415        logger.info("STTEngine prewarmed: model '%s' loaded", self.model_name)

Load the model eagerly (avoids cold-start latency on first transcription).

def close(self) -> None: View Source

417    def close(self) -> None:
418        """Release model resources."""
419        with self._model_lock:
420            self._model = None

Release model resources.

@dataclass

class StreamingSTTEngine: View Source

445@dataclass
446class StreamingSTTEngine:
447    """Incremental streaming STT: accepts audio chunks, yields segments.
448
449    Audio chunks are pushed one at a time via :meth:`push_chunk`.  When the
450    accumulated buffer reaches ``min_buffer_seconds``, :meth:`push_chunk`
451    transparently transcribes the buffer and yields any new segments.  You can
452    also force a transcription at any time with :meth:`flush`.
453
454    A sliding-window approach is supported via ``stride_seconds``: after each
455    transcription pass the engine retains the last ``stride_seconds`` of audio
456    so that words near the boundary are not lost on the next pass.  Set
457    ``stride_seconds=0.0`` (default) to discard all audio after each pass.
458
459    Thread safety: **not** thread-safe.  Call from a single thread or protect
460    externally with a lock.
461
462    Args:
463        model: Whisper model size. One of ``tiny``, ``base``, ``small``,
464               ``medium``, ``large-v3``. Default ``"base"``.
465        device: ``"cpu"`` or ``"cuda"``. Default ``"cpu"``.
466        compute_type: CTranslate2 compute type. Default ``"int8"``.
467        language: Force a specific language code (e.g. ``"en"``). ``None``
468                  for auto-detect.
469        min_buffer_seconds: Minimum seconds of audio to accumulate before
470                            attempting a transcription pass.  Shorter values
471                            mean lower latency but more frequent (and
472                            potentially noisier) passes.  Default ``2.0``.
473        stride_seconds: Seconds of audio overlap to retain between passes
474                        (sliding-window).  Default ``0.0`` (no overlap).
475        sample_rate: Sample rate of incoming audio chunks. Default ``16000``.
476
477    Example::
478
479        streaming = StreamingSTTEngine(model="base", min_buffer_seconds=2.0)
480        for chunk in mic_chunks:
481            for segment in streaming.push_chunk(chunk):
482                print(f"[{segment.start:.1f}s] {segment.text}")
483
484        # Force final transcription when done
485        for segment in streaming.flush():
486            print(f"[{segment.start:.1f}s] {segment.text}")
487    """
488
489    model: str = "base"
490    device: str = "cpu"
491    compute_type: str = "int8"
492    language: str | None = None
493    min_buffer_seconds: float = 2.0
494    stride_seconds: float = _DEFAULT_STRIDE_S
495    sample_rate: int = 16_000
496
497    # Internal state — populated post-init; not part of the public constructor.
498    _engine: STTEngine = field(init=False, repr=False)
499    _buffer: list[np.ndarray] = field(init=False, repr=False, default_factory=list)
500    _buffer_samples: int = field(init=False, repr=False, default=0)
501
502    def __post_init__(self) -> None:
503        self._engine = STTEngine(
504            model=self.model,
505            device=self.device,
506            compute_type=self.compute_type,
507            language=self.language,
508        )
509        self._buffer = []
510        self._buffer_samples = 0
511        logger.info(
512            "StreamingSTTEngine created: model=%s, min_buffer=%.1fs, stride=%.1fs",
513            self.model,
514            self.min_buffer_seconds,
515            self.stride_seconds,
516        )
517
518    # ------------------------------------------------------------------
519    # Public API
520    # ------------------------------------------------------------------
521
522    @property
523    def buffer_duration_s(self) -> float:
524        """Current accumulated audio duration in seconds."""
525        return self._buffer_samples / self.sample_rate
526
527    def push_chunk(self, chunk: np.ndarray | bytes) -> Iterator[TranscriptSegment]:
528        """Push an audio chunk into the buffer.
529
530        If the buffer has accumulated at least ``min_buffer_seconds`` of audio,
531        a transcription pass is run and any yielded segments are returned.
532        Otherwise, no segments are yielded and the chunk is silently buffered.
533
534        Args:
535            chunk: Float32 numpy array (16kHz mono) **or** raw ``int16`` PCM
536                   bytes.  Bytes are automatically converted to float32.
537
538        Yields:
539            TranscriptSegment — segments decoded in this pass (may be empty).
540        """
541        arr = self._coerce_chunk(chunk)
542        self._buffer.append(arr)
543        self._buffer_samples += len(arr)
544
545        min_samples = int(self.min_buffer_seconds * self.sample_rate)
546        if self._buffer_samples >= min_samples:
547            yield from self._run_pass()
548
549    def flush(self) -> Iterator[TranscriptSegment]:
550        """Transcribe whatever remains in the buffer and clear it.
551
552        Call this when the audio stream ends to ensure trailing audio is
553        transcribed.
554
555        Yields:
556            TranscriptSegment — segments from the remaining buffer.
557        """
558        if self._buffer_samples == 0:
559            return
560
561        logger.debug("StreamingSTTEngine.flush: %.2f s buffered", self.buffer_duration_s)
562        yield from self._run_pass(force=True)
563
564    def reset(self) -> None:
565        """Discard the current buffer without transcribing."""
566        self._buffer = []
567        self._buffer_samples = 0
568        logger.debug("StreamingSTTEngine buffer reset")
569
570    def prewarm(self) -> None:
571        """Eagerly load the underlying Whisper model."""
572        self._engine.prewarm()
573
574    def close(self) -> None:
575        """Release model resources and discard the buffer."""
576        self.reset()
577        self._engine.close()
578
579    def __enter__(self) -> StreamingSTTEngine:
580        """Enter sync context manager."""
581        return self
582
583    def __exit__(
584        self,
585        exc_type: type[BaseException] | None,
586        exc_val: BaseException | None,
587        exc_tb: object,
588    ) -> None:
589        """Exit sync context manager. Releases engine resources."""
590        self.close()
591
592    # ------------------------------------------------------------------
593    # Internal helpers
594    # ------------------------------------------------------------------
595
596    def _coerce_chunk(self, chunk: np.ndarray | bytes) -> np.ndarray:
597        """Convert raw int16 bytes or ensure float32 array."""
598        if isinstance(chunk, (bytes, bytearray)):
599            arr = np.frombuffer(chunk, dtype=np.int16).astype(np.float32) / 32768.0
600            return arr
601        arr = np.asarray(chunk, dtype=np.float32)
602        if arr.ndim > 1:
603            # Best-effort stereo → mono using the shape heuristic from STTEngine
604            arr = arr.mean(axis=0) if arr.shape[0] < arr.shape[1] else arr.mean(axis=1)
605        return arr
606
607    def _run_pass(self, *, force: bool = False) -> Iterator[TranscriptSegment]:
608        """Concatenate buffer, transcribe, apply sliding window, yield segments."""
609        audio = np.concatenate(self._buffer)
610
611        logger.debug(
612            "StreamingSTTEngine pass: %.2f s (force=%s)",
613            len(audio) / self.sample_rate,
614            force,
615        )
616
617        yield from self._engine.transcribe_streaming(audio)
618
619        # Sliding window: retain the last stride_seconds of audio so that
620        # words near the boundary are not cut off on the next pass.
621        stride_samples = int(self.stride_seconds * self.sample_rate)
622        if stride_samples > 0 and len(audio) > stride_samples:
623            retained = audio[-stride_samples:]
624            self._buffer = [retained]
625            self._buffer_samples = len(retained)
626        else:
627            self._buffer = []
628            self._buffer_samples = 0

Incremental streaming STT: accepts audio chunks, yields segments.

Audio chunks are pushed one at a time via push_chunk(). When the accumulated buffer reaches min_buffer_seconds, push_chunk() transparently transcribes the buffer and yields any new segments. You can also force a transcription at any time with flush().

A sliding-window approach is supported via stride_seconds: after each transcription pass the engine retains the last stride_seconds of audio so that words near the boundary are not lost on the next pass. Set stride_seconds=0.0 (default) to discard all audio after each pass.

Thread safety: not thread-safe. Call from a single thread or protect externally with a lock.

Arguments:

model: Whisper model size. One of tiny, base, small, medium, large-v3. Default "base".
device: "cpu" or "cuda". Default "cpu".
compute_type: CTranslate2 compute type. Default "int8".
language: Force a specific language code (e.g. "en"). None for auto-detect.
min_buffer_seconds: Minimum seconds of audio to accumulate before attempting a transcription pass. Shorter values mean lower latency but more frequent (and potentially noisier) passes. Default 2.0.
stride_seconds: Seconds of audio overlap to retain between passes (sliding-window). Default 0.0 (no overlap).
sample_rate: Sample rate of incoming audio chunks. Default 16000.

Example::

streaming = StreamingSTTEngine(model="base", min_buffer_seconds=2.0)
for chunk in mic_chunks:
    for segment in streaming.push_chunk(chunk):
        print(f"[{segment.start:.1f}s] {segment.text}")

# Force final transcription when done
for segment in streaming.flush():
    print(f"[{segment.start:.1f}s] {segment.text}")

StreamingSTTEngine( model: str = 'base', device: str = 'cpu', compute_type: str = 'int8', language: str | None = None, min_buffer_seconds: float = 2.0, stride_seconds: float = 0.0, sample_rate: int = 16000)

model: str = 'base'

device: str = 'cpu'

compute_type: str = 'int8'

language: str | None = None

min_buffer_seconds: float = 2.0

stride_seconds: float = 0.0

sample_rate: int = 16000

buffer_duration_s: float View Source

522    @property
523    def buffer_duration_s(self) -> float:
524        """Current accumulated audio duration in seconds."""
525        return self._buffer_samples / self.sample_rate

Current accumulated audio duration in seconds.

def push_chunk( self, chunk: numpy.ndarray | bytes) -> Iterator[violawake_sdk.stt.TranscriptSegment]: View Source

527    def push_chunk(self, chunk: np.ndarray | bytes) -> Iterator[TranscriptSegment]:
528        """Push an audio chunk into the buffer.
529
530        If the buffer has accumulated at least ``min_buffer_seconds`` of audio,
531        a transcription pass is run and any yielded segments are returned.
532        Otherwise, no segments are yielded and the chunk is silently buffered.
533
534        Args:
535            chunk: Float32 numpy array (16kHz mono) **or** raw ``int16`` PCM
536                   bytes.  Bytes are automatically converted to float32.
537
538        Yields:
539            TranscriptSegment — segments decoded in this pass (may be empty).
540        """
541        arr = self._coerce_chunk(chunk)
542        self._buffer.append(arr)
543        self._buffer_samples += len(arr)
544
545        min_samples = int(self.min_buffer_seconds * self.sample_rate)
546        if self._buffer_samples >= min_samples:
547            yield from self._run_pass()

Push an audio chunk into the buffer.

If the buffer has accumulated at least min_buffer_seconds of audio, a transcription pass is run and any yielded segments are returned. Otherwise, no segments are yielded and the chunk is silently buffered.

Arguments:

chunk: Float32 numpy array (16kHz mono) or raw int16 PCM bytes. Bytes are automatically converted to float32.

Yields:

TranscriptSegment — segments decoded in this pass (may be empty).

def flush(self) -> Iterator[violawake_sdk.stt.TranscriptSegment]: View Source

549    def flush(self) -> Iterator[TranscriptSegment]:
550        """Transcribe whatever remains in the buffer and clear it.
551
552        Call this when the audio stream ends to ensure trailing audio is
553        transcribed.
554
555        Yields:
556            TranscriptSegment — segments from the remaining buffer.
557        """
558        if self._buffer_samples == 0:
559            return
560
561        logger.debug("StreamingSTTEngine.flush: %.2f s buffered", self.buffer_duration_s)
562        yield from self._run_pass(force=True)

Transcribe whatever remains in the buffer and clear it.

Call this when the audio stream ends to ensure trailing audio is transcribed.

Yields:

TranscriptSegment — segments from the remaining buffer.

def reset(self) -> None: View Source

564    def reset(self) -> None:
565        """Discard the current buffer without transcribing."""
566        self._buffer = []
567        self._buffer_samples = 0
568        logger.debug("StreamingSTTEngine buffer reset")

Discard the current buffer without transcribing.

def prewarm(self) -> None: View Source

570    def prewarm(self) -> None:
571        """Eagerly load the underlying Whisper model."""
572        self._engine.prewarm()

Eagerly load the underlying Whisper model.

def close(self) -> None: View Source

574    def close(self) -> None:
575        """Release model resources and discard the buffer."""
576        self.reset()
577        self._engine.close()

Release model resources and discard the buffer.

class ViolaWakeError(builtins.Exception): View Source

7class ViolaWakeError(Exception):
8    """Base exception for all ViolaWake SDK errors."""

Base exception for all ViolaWake SDK errors.

class ModelNotFoundError(violawake_sdk.ViolaWakeError): View Source

11class ModelNotFoundError(ViolaWakeError):
12    """Raised when a model file is not found in the cache or at the given path.
13
14    Resolution: run ``violawake-download --model <model_name>`` to download.
15    """

Raised when a model file is not found in the cache or at the given path.

Resolution: run violawake-download --model <model_name> to download.

class AudioCaptureError(violawake_sdk.ViolaWakeError): View Source

25class AudioCaptureError(ViolaWakeError):
26    """Raised when microphone capture fails to initialize or read frames.
27
28    Common causes: no audio input device, device already in use,
29    PortAudio not installed.
30    """

Raised when microphone capture fails to initialize or read frames.

Common causes: no audio input device, device already in use, PortAudio not installed.

class ModelLoadError(violawake_sdk.ViolaWakeError): View Source

18class ModelLoadError(ViolaWakeError):
19    """Raised when a model file exists but cannot be loaded by ONNX Runtime.
20
21    Possible causes: corrupted file, ONNX opset version mismatch.
22    """

Raised when a model file exists but cannot be loaded by ONNX Runtime.

Possible causes: corrupted file, ONNX opset version mismatch.

class PipelineError(violawake_sdk.ViolaWakeError): View Source

40class PipelineError(ViolaWakeError):
41    """Raised when the VoicePipeline encounters an unrecoverable error."""

Raised when the VoicePipeline encounters an unrecoverable error.

class VADBackendError(violawake_sdk.ViolaWakeError): View Source

33class VADBackendError(ViolaWakeError):
34    """Raised when the requested VAD backend is unavailable.
35
36    Falls back to RMS heuristic if webrtcvad/silero not installed.
37    """

Raised when the requested VAD backend is unavailable.

Falls back to RMS heuristic if webrtcvad/silero not installed.

def list_models() -> list[dict[str, str]]: View Source

104def list_models() -> list[dict[str, str]]:
105    """Return available wake word models with their descriptions.
106
107    Each entry is a dict with keys: ``name``, ``description``, ``version``.
108
109    Example::
110
111        >>> from violawake_sdk import list_models
112        >>> for m in list_models():
113        ...     print(f"{m['name']:20s} {m['description']}")
114    """
115    from violawake_sdk.models import MODEL_REGISTRY
116
117    seen: set[str] = set()
118    result: list[dict[str, str]] = []
119    for name, spec in MODEL_REGISTRY.items():
120        # Deduplicate aliases (e.g. "viola" -> "temporal_cnn")
121        if spec.name in seen:
122            continue
123        # Hide deprecated, package-managed, and non-wake-word models
124        if "DEPRECATED" in spec.description:
125            continue
126        if spec.name in ("oww_backbone", "kokoro_v1_0", "kokoro_voices_v1_0"):
127            continue
128        seen.add(spec.name)
129        result.append(
130            {
131                "name": name,
132                "description": spec.description,
133                "version": spec.version,
134            }
135        )
136    return result

Return available wake word models with their descriptions.

Each entry is a dict with keys: name, description, version.

Example::

>>> from violawake_sdk import list_models
>>> for m in list_models():
...     print(f"{m['name']:20s} {m['description']}")

def list_voices() -> list[str]: View Source

139def list_voices() -> list[str]:
140    """Return available TTS voice names for use with ``TTSEngine``.
141
142    Requires the ``[tts]`` extra to be installed for actual synthesis,
143    but this function always works for discovery.
144
145    Example::
146
147        >>> from violawake_sdk import list_voices
148        >>> list_voices()
149        ['af_heart', 'af_bella', 'af_sarah', ...]
150    """
151    from violawake_sdk.tts import AVAILABLE_VOICES
152
153    return list(AVAILABLE_VOICES)

Return available TTS voice names for use with TTSEngine.

Requires the [tts] extra to be installed for actual synthesis, but this function always works for discovery.

Example::

>>> from violawake_sdk import list_voices
>>> list_voices()
['af_heart', 'af_bella', 'af_sarah', ...]

__version__ = '0.2.9'