From 9bc2830a2de4d4843ce0674135bf9bc02a71154f Mon Sep 17 00:00:00 2001 From: imonlinux <39863321+imonlinux@users.noreply.github.com> Date: Mon, 17 Nov 2025 20:12:18 -0800 Subject: [PATCH] Delete linux_voice_assistant/microwakeword.py --- linux_voice_assistant/microwakeword.py | 189 ------------------------- 1 file changed, 189 deletions(-) delete mode 100644 linux_voice_assistant/microwakeword.py diff --git a/linux_voice_assistant/microwakeword.py b/linux_voice_assistant/microwakeword.py deleted file mode 100644 index 79b4067..0000000 --- a/linux_voice_assistant/microwakeword.py +++ /dev/null @@ -1,189 +0,0 @@ -import ctypes -import json -import statistics -from collections import deque -from collections.abc import Iterable -from pathlib import Path -from typing import Deque, List, Union - -import numpy as np -from pymicro_features import MicroFrontend - -from .wakeword import TfLiteWakeWord - -SAMPLES_PER_SECOND = 16000 -SAMPLES_PER_CHUNK = 160 # 10ms -BYTES_PER_SAMPLE = 2 # 16-bit -BYTES_PER_CHUNK = SAMPLES_PER_CHUNK * BYTES_PER_SAMPLE -SECONDS_PER_CHUNK = SAMPLES_PER_CHUNK / SAMPLES_PER_SECOND -STRIDE = 3 - - -class MicroWakeWord(TfLiteWakeWord): - def __init__( - self, - id: str, # pylint: disable=redefined-builtin - wake_word: str, - tflite_model: Union[str, Path], - probability_cutoff: float, - sliding_window_size: int, - trained_languages: List[str], - libtensorflowlite_c_path: Union[str, Path], - ) -> None: - TfLiteWakeWord.__init__(self, libtensorflowlite_c_path) - - self.id = id - self.wake_word = wake_word - self.tflite_model = tflite_model - self.probability_cutoff = probability_cutoff - self.sliding_window_size = sliding_window_size - self.trained_languages = trained_languages - - self.is_active = True - - # Load the model and create interpreter - self.model_path = str(Path(tflite_model).resolve()).encode("utf-8") - self._load_model() - - self._features: List[np.ndarray] = [] - self._probabilities: Deque[float] = deque(maxlen=self.sliding_window_size) - self._audio_buffer = bytes() - - def _load_model(self) -> None: - self.model = self.lib.TfLiteModelCreateFromFile(self.model_path) - self.interpreter = self.lib.TfLiteInterpreterCreate(self.model, None) - self.lib.TfLiteInterpreterAllocateTensors(self.interpreter) - - # Access input and output tensor - self.input_tensor = self.lib.TfLiteInterpreterGetInputTensor( - self.interpreter, 0 - ) - self.output_tensor = self.lib.TfLiteInterpreterGetOutputTensor( - self.interpreter, 0 - ) - - # Get quantization parameters - input_q = self.lib.TfLiteTensorQuantizationParams(self.input_tensor) - output_q = self.lib.TfLiteTensorQuantizationParams(self.output_tensor) - - self.input_scale, self.input_zero_point = input_q.scale, input_q.zero_point - self.output_scale, self.output_zero_point = output_q.scale, output_q.zero_point - - @staticmethod - def from_config( - config_path: Union[str, Path], - libtensorflowlite_c_path: Union[str, Path], - ) -> "MicroWakeWord": - """Load a microWakeWord model from a JSON config file. - - Parameters - ---------- - config_path: str or Path - Path to JSON configuration file - """ - config_path = Path(config_path) - with open(config_path, "r", encoding="utf-8") as config_file: - config = json.load(config_file) - - micro_config = config["micro"] - - return MicroWakeWord( - id=Path(config["model"]).stem, - wake_word=config["wake_word"], - tflite_model=config_path.parent / config["model"], - probability_cutoff=micro_config["probability_cutoff"], - sliding_window_size=micro_config["sliding_window_size"], - trained_languages=micro_config.get("trained_languages", []), - libtensorflowlite_c_path=libtensorflowlite_c_path, - ) - - def process_streaming(self, features: np.ndarray) -> bool: - self._features.append(features) - - if len(self._features) < STRIDE: - # Not enough windows - return False - - # Allocate and quantize input data - quant_features = np.round( - np.concatenate(self._features, axis=1) / self.input_scale - + self.input_zero_point - ).astype(np.uint8) - - # Stride instead of rolling - self._features.clear() - - # Set tensor - quant_ptr = quant_features.ctypes.data_as(ctypes.c_void_p) - self.lib.TfLiteTensorCopyFromBuffer( - self.input_tensor, quant_ptr, quant_features.nbytes - ) - - # Run inference - self.lib.TfLiteInterpreterInvoke(self.interpreter) - - # Read output - output_bytes = self.lib.TfLiteTensorByteSize(self.output_tensor) - output_data = np.empty(output_bytes, dtype=np.uint8) - self.lib.TfLiteTensorCopyToBuffer( - self.output_tensor, - output_data.ctypes.data_as(ctypes.c_void_p), - output_bytes, - ) - - # Dequantize output - result = ( - output_data.astype(np.float32) - self.output_zero_point - ) * self.output_scale - - self._probabilities.append(result.item()) - - if len(self._probabilities) < self.sliding_window_size: - # Not enough probabilities - return False - - if statistics.mean(self._probabilities) > self.probability_cutoff: - return True - - return False - - -# ----------------------------------------------------------------------------- - - -class MicroWakeWordFeatures(TfLiteWakeWord): - def __init__( - self, - libtensorflowlite_c_path: Union[str, Path], - ) -> None: - TfLiteWakeWord.__init__(self, libtensorflowlite_c_path) - - self._audio_buffer = bytes() - self._frontend = MicroFrontend() - - def process_streaming(self, audio_bytes: bytes) -> Iterable[np.ndarray]: - self._audio_buffer += audio_bytes - - if len(self._audio_buffer) < BYTES_PER_CHUNK: - # Not enough audio to get features - return - - audio_buffer_idx = 0 - while (audio_buffer_idx + BYTES_PER_CHUNK) <= len(self._audio_buffer): - # Process chunk - chunk_bytes = self._audio_buffer[ - audio_buffer_idx : audio_buffer_idx + BYTES_PER_CHUNK - ] - frontend_result = self._frontend.ProcessSamples(chunk_bytes) - audio_buffer_idx += frontend_result.samples_read * BYTES_PER_SAMPLE - - if not frontend_result.features: - # Not enough audio for a full window - continue - - yield np.array(frontend_result.features).reshape( - (1, 1, len(frontend_result.features)) - ) - - # Remove processed audio - self._audio_buffer = self._audio_buffer[audio_buffer_idx:]