cjm-media-plugin-silero-vad 0.0.22

A local, persistent Voice Activity Detection (VAD) worker for the cjm-plugin-system that provides high-accuracy speech segmentation using Silero VAD with SQLite result caching.

cjm-media-plugin-silero-vad

Install

pip install cjm_media_plugin_silero_vad

Project Structure

nbs/
└── plugin.ipynb # Pure-compute voice-activity-detection tool capability using Silero VAD (Option C, stage 8).

Total: 1 notebook

Module Dependencies

graph LR
    plugin["plugin<br/>Silero VAD Plugin"]

No cross-module dependencies detected.

CLI Reference

No CLI commands found in this project.

Module Overview

Detailed documentation for each module in the project:

Silero VAD Plugin (plugin.ipynb)

Pure-compute voice-activity-detection tool capability using Silero VAD (Option C, stage 8).

Import

from cjm_media_plugin_silero_vad.plugin import (
    SileroVADConfig,
    SileroVADPlugin
)

Classes

@dataclass
class SileroVADConfig:
    "Configuration for Silero VAD parameters."
    
    threshold: float = field(...)
    min_speech_duration_ms: int = field(...)
    min_silence_duration_ms: int = field(...)
    speech_pad_ms: int = field(...)
    use_onnx: bool = field(...)

class SileroVADPlugin:
    def __init__(self):
        """Initialize the Silero VAD plugin."""
        self.logger = logging.getLogger(f"{__name__}.{type(self).__name__}")
        self.config: SileroVADConfig = None
    """
    Voice Activity Detection tool capability using Silero VAD (stage 8: pure compute).
    
    Native-surface model (PILLAR 1c): this tool is PURE COMPUTE — `detect_speech`
    reads MODEL-READY audio, runs Silero inference, and builds the typed
    `VADResult`. The cache-check + persistence bookends + the per-call `force`
    control live in the generic VAD adapter (cjm-vad-adapter-interface); the
    result DTO lives in cjm-capability-primitives; identity is derived from the
    installed distribution. No `get_plugin_metadata`, no `self.storage`, no
    librosa (decode/resample is upstream ffmpeg, soxr).
    """
    
    def __init__(self):
            """Initialize the Silero VAD plugin."""
            self.logger = logging.getLogger(f"{__name__}.{type(self).__name__}")
            self.config: SileroVADConfig = None
        "Initialize the Silero VAD plugin."
    
    def name(self) -> str:  # Plugin name identifier
            """Plugin identity, derived from the installed distribution (PILLAR 1c)."""
            from importlib.metadata import metadata, packages_distributions
            dist = (packages_distributions().get(__package__) or [__package__.replace("_", "-")])[0]
            return metadata(dist)["Name"]
    
        @property
        def version(self) -> str:  # Plugin version string
        "Plugin identity, derived from the installed distribution (PILLAR 1c)."
    
    def version(self) -> str:  # Plugin version string
            """Get the plugin version string."""
            from cjm_media_plugin_silero_vad import __version__
            return __version__
    
        def get_current_config(self) -> Dict[str, Any]:  # Current configuration as dictionary
        "Get the plugin version string."
    
    def get_current_config(self) -> Dict[str, Any]:  # Current configuration as dictionary
            """Return current configuration state."""
            return config_to_dict(self.config) if self.config else {}
    
        def get_config_schema(self) -> Dict[str, Any]:  # JSON Schema for configuration
        "Return current configuration state."
    
    def get_config_schema(self) -> Dict[str, Any]:  # JSON Schema for configuration
            """Return JSON Schema for UI generation."""
            return dataclass_to_jsonschema(SileroVADConfig)
    
        def _apply_config(
            self,
            config: Optional[Any] = None  # Configuration dataclass, dict, or None
        ) -> None
        "Return JSON Schema for UI generation."
    
    def initialize(
            self,
            config: Optional[Any] = None  # Configuration dataclass, dict, or None
        ) -> None
        "First-time setup. CR-4: config application is factored into _apply_config;
the substrate's reconfigure(old, new) fires _release_model on a use_onnx
change (RELOAD_TRIGGER) then re-applies config. No storage init — the
adapter owns the cache (stage 8)."
    
    def detect_speech(
            self,
            audio: Union[str, Path],  # Path to MODEL-READY audio (mono 8k/16k, converted upstream)
            **kwargs                  # Provenance pass-through (unused by VAD compute)
        ) -> VADResult:  # Typed VAD output with detected speech segments
        "Detect speech segments in model-ready audio — PURE COMPUTE.

Stage 8 / PILLAR 1c: the cache-check + persistence bookends + the per-call
`force` control moved to the generic VAD adapter; this method reads the
audio, runs Silero, and builds the typed result. Detection params come
from `self.config` (no per-call kwarg override — the tool runs its
effective config)."
    
    def is_available(self) -> bool:  # True if Silero VAD is available
            """Check if Silero VAD is available."""
            return SILERO_AVAILABLE
    
        def prefetch(self) -> None
        "Check if Silero VAD is available."
    
    def prefetch(self) -> None:
            """CR-4 (SG-19): eagerly load the model so the first call doesn't pay the load cost."""
            self._load_model()
    
        def on_disable(self) -> None
        "CR-4 (SG-19): eagerly load the model so the first call doesn't pay the load cost."
    
    def on_disable(self) -> None:
            """CR-2: release the model when the operator disables the plugin (worker stays alive)."""
            self._release_model()
    
        def cleanup(self) -> None
        "CR-2: release the model when the operator disables the plugin (worker stays alive)."
    
    def cleanup(self) -> None
        "Release resources on unload."