Allow remotely triggered pipelines for wake word satellites

wyoming_satellite/vad.py

@@ -1,6 +1,10 @@
 """Voice activity detection."""
+import logging
+from dataclasses import dataclass
 from typing import Optional
 
+_LOGGER = logging.getLogger(__name__)
+
 
 class SileroVad:
     """Voice activity detection with silero VAD."""
@@ -31,3 +35,95 @@ class SileroVad:
             self._activation = max(0, self._activation - 1)
 
         return False
+
+
+@dataclass
+class VoiceCommandSegmenter:
+    """Segments an audio stream into voice commands."""
+
+    speech_seconds: float = 0.3
+    """Seconds of speech before voice command has started."""
+
+    silence_seconds: float = 1.0
+    """Seconds of silence after voice command has ended."""
+
+    timeout_seconds: float = 15.0
+    """Maximum number of seconds before stopping with timeout=True."""
+
+    reset_seconds: float = 1.0
+    """Seconds before reset start/stop time counters."""
+
+    in_command: bool = False
+    """True if inside voice command."""
+
+    _speech_seconds_left: float = 0.0
+    """Seconds left before considering voice command as started."""
+
+    _silence_seconds_left: float = 0.0
+    """Seconds left before considering voice command as stopped."""
+
+    _timeout_seconds_left: float = 0.0
+    """Seconds left before considering voice command timed out."""
+
+    _reset_seconds_left: float = 0.0
+    """Seconds left before resetting start/stop time counters."""
+
+    def __post_init__(self) -> None:
+        """Reset after initialization."""
+        self.reset()
+
+    def reset(self) -> None:
+        """Reset all counters and state."""
+        self._speech_seconds_left = self.speech_seconds
+        self._silence_seconds_left = self.silence_seconds
+        self._timeout_seconds_left = self.timeout_seconds
+        self._reset_seconds_left = self.reset_seconds
+        self.in_command = False
+
+    def process(self, chunk_seconds: float, is_speech: bool | None) -> bool:
+        """Process samples using external VAD.
+
+        Returns False when command is done.
+        """
+        self._timeout_seconds_left -= chunk_seconds
+        if self._timeout_seconds_left <= 0:
+            _LOGGER.warning(
+                "VAD end of speech detection timed out after %s seconds",
+                self.timeout_seconds,
+            )
+            self.reset()
+            return False
+
+        if not self.in_command:
+            if is_speech:
+                self._reset_seconds_left = self.reset_seconds
+                self._speech_seconds_left -= chunk_seconds
+                if self._speech_seconds_left <= 0:
+                    # Inside voice command
+                    self.in_command = True
+                    self._silence_seconds_left = self.silence_seconds
+                    _LOGGER.debug("Voice command started")
+            else:
+                # Reset if enough silence
+                self._reset_seconds_left -= chunk_seconds
+                if self._reset_seconds_left <= 0:
+                    self._speech_seconds_left = self.speech_seconds
+                    self._reset_seconds_left = self.reset_seconds
+        elif not is_speech:
+            # Silence in command
+            self._reset_seconds_left = self.reset_seconds
+            self._silence_seconds_left -= chunk_seconds
+            if self._silence_seconds_left <= 0:
+                # Command finished successfully
+                self.reset()
+                _LOGGER.debug("Voice command finished")
+                return False
+        else:
+            # Speech in command.
+            # Reset silence counter if enough speech.
+            self._reset_seconds_left -= chunk_seconds
+            if self._reset_seconds_left <= 0:
+                self._silence_seconds_left = self.silence_seconds
+                self._reset_seconds_left = self.reset_seconds
+
+        return True