import sounddevice as sd
import numpy as np
from collections import deque
import threading

class AudioRingBuffer:
    def __init__(self, sample_rate: int = 16000, duration_sec: int = 5):
        self.sample_rate = sample_rate
        self.max_samples = sample_rate * duration_sec
        self._buffer = deque(maxlen=self.max_samples)
        self._lock = threading.Lock()
        self._stream = None

    def _callback(self, indata: np.ndarray, _frames, _time, _status):
        with self._lock:
            self._buffer.extend(indata[:, 0])

    def start(self):
        self._stream = sd.InputStream(
            samplerate=self.sample_rate,
            channels=1,
            dtype=np.float32,
            callback=self._callback
        )
        self._stream.start()

    def get_slice(self, duration_sec: float) -> np.ndarray:
        with self._lock:
            required_samples = int(self.sample_rate * duration_sec)
            if len(self._buffer) < required_samples:
                return np.array([])
            return np.array(list(self._buffer)[-required_samples:])

    def stop(self):
        if self._stream:
            self._stream.stop()
            self._stream.close()

import tflite_runtime.interpreter as tflite
import numpy as np

class IntentClassifier:
    def __init__(self, model_path: str, intent_labels: dict):
        self.interpreter = tflite.Interpreter(model_path=model_path)
        self.interpreter.allocate_tensors()
        self.input_details = self.interpreter.get_input_details()[0]
        self.output_details = self.interpreter.get_output_details()[0]
        self.labels = intent_labels

    def classify(self, waveform: np.ndarray) -> tuple[str, float]:
        # Quantized models expect int16 input in this architecture
        input_data = waveform.astype(np.int16).reshape(1, -1, 1)
        self.interpreter.set_tensor(self.input_details["index"], input_data)
        self.interpreter.invoke()
        probabilities = self.interpreter.get_tensor(self.output_details["index"])[0]
        predicted_idx = int(np.argmax(probabilities))
        confidence = float(probabilities[predicted_idx])
        return self.labels[predicted_idx], confidence

es dependency footprint by ~80%. Post-training quantization (tf.lite.Optimize.DEFAULT) converts float32 weights to INT8 without retraining, preserving accuracy while enabling NEON SIMD acceleration on Cortex-A cores. The classifier operates on 1-second windows to minimize latency.

3. Periodic Transcription Engine (Whisper)

Full transcription runs asynchronously on 2-second windows. We constrain thread usage and disable beam search to prioritize speed over marginal accuracy gains.

import whisper
import torch
import numpy as np

class TranscriptionEngine:
    def __init__(self, model_name: str = "small", device: str = "cpu"):
        torch.set_num_threads(2)
        self.model = whisper.load_model(model_name, device=device)

    def transcribe(self, audio_chunk: np.ndarray) -> str:
        if len(audio_chunk) == 0:
            return ""
        tensor_input = torch.from_numpy(audio_chunk).float()
        result = self.model.transcribe(
            tensor_input,
            language="en",
            word_timestamps=False,
            beam_size=1
        )
        return result["text"].strip()

Architecture Rationale: beam_size=1 switches Whisper from beam search to greedy decoding, reducing compute by ~40% with minimal accuracy loss for short commands. Limiting PyTorch to 2 threads prevents CPU starvation when the intent classifier and audio callback are active. The engine runs every 2 seconds, providing contextual logs without blocking the fast path.

4. Command Router & Execution Loop

The main loop polls both inference paths, applies confidence gating, and dispatches actions.

import time
import logging

class VoiceCommandRouter:
    def __init__(self, audio_buffer: AudioRingBuffer, 
                 intent_classifier: IntentClassifier, 
                 transcriber: TranscriptionEngine,
                 confidence_threshold: float = 0.85):
        self.audio = audio_buffer
        self.classifier = intent_classifier
        self.transcriber = transcriber
        self.threshold = confidence_threshold
        logging.basicConfig(level=logging.INFO)

    def run(self):
        self.audio.start()
        logging.info("Voice pipeline initialized. Awaiting input...")
        try:
            while True:
                # Fast path: intent detection
                intent_window = self.audio.get_slice(1.0)
                if len(intent_window) > 0:
                    label, conf = self.classifier.classify(intent_window)
                    if conf >= self.threshold:
                        self._dispatch(label)

                # Slow path: full transcription
                transcribe_window = self.audio.get_slice(2.0)
                if len(transcribe_window) > 0:
                    text = self.transcriber.transcribe(transcribe_window)
                    if text:
                        logging.info(f"[TRANSCRIPT] {text}")

                time.sleep(0.15)
        except KeyboardInterrupt:
            logging.info("Pipeline halted by user.")
        finally:
            self.audio.stop()

    def _dispatch(self, intent: str):
        actions = {
            "ACTIVATE_LIGHT": lambda: logging.info("Relay: ON"),
            "DEACTIVATE_LIGHT": lambda: logging.info("Relay: OFF"),
            "QUERY_TIME": lambda: logging.info(f"System time: {time.strftime('%H:%M')}"),
            "TERMINATE": lambda: (_ for _ in ()).throw(KeyboardInterrupt)
        }
        handler = actions.get(intent)
        if handler:
            handler()
        else:
            logging.warning(f"Unmapped intent: {intent}")

Architecture Rationale: Separating routing from inference improves testability and allows hot-swapping models. Confidence gating prevents false triggers from background noise. The 150 ms sleep interval balances CPU utilization with responsiveness. Actions are mapped to lambdas for clean extension without modifying core logic.

Pitfall Guide

Pitfall	Explanation	Fix
Blocking the audio callback	Performing inference or I/O inside the sounddevice callback causes buffer underruns and audio dropouts.	Keep the callback strictly limited to buffer writes. Run all inference in the main thread or worker threads.
Sample rate mismatch	Whisper and TFLite models expect 16 kHz. Feeding 44.1 kHz or 48 kHz audio degrades accuracy and increases compute.	Configure sd.InputStream with samplerate=16000. Validate input shapes before inference.
Skipping post-training quantization	Float32 TFLite models consume 4x memory and run 3–5x slower on ARM CPUs without NEON optimization.	Apply tf.lite.Optimize.DEFAULT during conversion. Verify INT8 input/output signatures match your pipeline.
Hardcoding confidence thresholds	Fixed thresholds fail across different microphones, ambient noise levels, and speaker distances.	Implement adaptive thresholding or expose the value via environment configuration. Log false positives/negatives for tuning.
Thread pool contention	PyTorch defaults to using all available cores, starving the audio thread and intent classifier.	Call torch.set_num_threads(2) before loading Whisper. Pin critical threads to specific cores using taskset if needed.
Buffer overflow under load	If inference takes longer than the audio capture rate, the ring buffer may drop samples or cause memory pressure.	Monitor len(buffer) vs maxlen. Implement backpressure by skipping transcription frames when CPU load exceeds 85%.
Missing system dependencies	Whisper requires ffmpeg for audio decoding. sounddevice requires libportaudio2. Omitting these causes silent failures.	Document OS-level dependencies. Use apt install ffmpeg libportaudio2-dev in deployment scripts. Validate with ffmpeg -version at startup.

Production Bundle

Action Checklist

• Verify hardware: Raspberry Pi 4 (2 GB+ RAM), USB microphone, 64-bit Raspberry Pi OS
• Install system dependencies: ffmpeg, libportaudio2-dev, python3-venv
• Create isolated Python environment and install whisper, tflite-runtime, sounddevice, numpy
• Quantize intent classifier to INT8 and validate inference speed >100/sec on target hardware
• Configure torch.set_num_threads(2) and disable beam search in Whisper initialization
• Implement confidence thresholding with logging for false trigger analysis
• Package as systemd service with Restart=on-failure and StandardOutput=journal
• Test under network disconnect and high ambient noise to validate offline resilience

Decision Matrix

Scenario	Recommended Approach	Why	Cost Impact
Air-gapped industrial control	Local dual-path (Whisper-small + TFLite)	Zero network dependency, deterministic latency, full data control	Higher upfront hardware cost, $0 recurring
Consumer smart home with reliable WiFi	Cloud STT + local intent fallback	Leverages cloud accuracy for complex queries, edge handles critical commands	Moderate API costs, reduced edge compute
Memory-constrained Pi Zero 2W	Whisper-tiny + ONNX Runtime	Lower RAM footprint (~800 MB), ONNX offers better ARM optimization than TFLite on some builds	Slightly lower accuracy, requires model conversion pipeline
High-noise factory floor	Add VAD pre-filter + adaptive threshold	Voice Activity Detection reduces false triggers from machinery noise	Adds ~5 ms latency, requires VAD model integration

Configuration Template

# /etc/systemd/system/voice-pipeline.service
[Unit]
Description=Edge Voice Command Pipeline
After=network.target sound.target
Wants=sound.target

[Service]
Type=simple
User=pi
WorkingDirectory=/opt/voice-pipeline
ExecStart=/opt/voice-pipeline/venv/bin/python3 -m pipeline.runner
Restart=on-failure
RestartSec=5
Environment=PYTHONUNBUFFERED=1
Environment=WHISPER_MODEL=small
Environment=INTENT_THRESHOLD=0.85
StandardOutput=journal
StandardError=journal
# Hardening
NoNewPrivileges=true
ProtectSystem=strict
ReadWritePaths=/opt/voice-pipeline/logs

[Install]
WantedBy=multi-user.target

# config/env_loader.py
import os
from dataclasses import dataclass

@dataclass
class PipelineConfig:
    sample_rate: int = 16000
    whisper_model: str = os.getenv("WHISPER_MODEL", "small")
    intent_threshold: float = float(os.getenv("INTENT_THRESHOLD", "0.85"))
    transcription_interval: float = 2.0
    intent_window: float = 1.0
    buffer_duration: int = 5
    torch_threads: int = 2

    def validate(self):
        assert self.whisper_model in ("tiny", "base", "small"), "Unsupported Whisper variant"
        assert 0.0 <= self.intent_threshold <= 1.0, "Threshold must be probability range"
        return self

Quick Start Guide

1. Provision the environment: Flash Raspberry Pi OS 64-bit, connect a USB microphone, and run sudo apt update && sudo apt install -y ffmpeg libportaudio2-dev python3-venv.
2. Initialize the project: Create a virtual environment, install dependencies (pip install whisper tflite-runtime sounddevice numpy), and place your quantized intent_classifier.tflite in the project root.
3. Launch the pipeline: Execute the runner script. The system will initialize the audio stream, load Whisper-small, and begin polling for commands. Monitor logs via journalctl -u voice-pipeline -f.
4. Validate & tune: Speak test commands. Adjust INTENT_THRESHOLD in the environment file if false triggers occur. Verify CPU usage stays below 70% using htop before enabling systemd auto-start.