alignment

Alignment service for temporal coordination via Silero VAD plugin

AlignmentService

This service wraps the Silero VAD plugin to provide voice activity detection data. It processes audio files to extract time ranges where speech is detected, which are then used to align text segments with their corresponding audio positions.

source

AlignmentService


def AlignmentService(
    plugin_manager:PluginManager, # Plugin manager for accessing VAD plugin
    plugin_name:str='cjm-media-plugin-silero-vad', # Name of the VAD plugin
):

Service for temporal alignment via Silero VAD plugin.

Alignment Helpers

source

check_alignment_ready


def check_alignment_ready(
    segment_count:int, # Number of text segments
    chunk_count:int, # Number of VAD chunks
)->bool: # True if counts match and alignment can proceed

Check if segment and VAD chunk counts match for 1:1 alignment.

Tests

The following cells demonstrate the alignment service and helper functions.

# Test check_alignment_ready
print(f"3 segments vs 3 chunks: {check_alignment_ready(3, 3)}")  # True
print(f"3 segments vs 5 chunks: {check_alignment_ready(3, 5)}")  # False
print(f"0 segments vs 3 chunks: {check_alignment_ready(0, 3)}")  # False
print(f"3 segments vs 0 chunks: {check_alignment_ready(3, 0)}")  # False

3 segments vs 3 chunks: True
3 segments vs 5 chunks: False
0 segments vs 3 chunks: False
3 segments vs 0 chunks: False

AlignmentService with Silero VAD Plugin

These tests require the Silero VAD plugin to be installed and discoverable.

# Test AlignmentService with Silero VAD plugin
from pathlib import Path
from cjm_plugin_system.core.manager import PluginManager

# Calculate project root from notebook location (nbs/services/ -> project root)
project_root = Path.cwd().parent.parent
manifests_dir = project_root / ".cjm" / "manifests"

# Create plugin manager with explicit search path
manager = PluginManager(search_paths=[manifests_dir])
manager.discover_manifests()

print(f"Discovered {len(manager.discovered)} plugins from {manifests_dir}")

# Check if VAD plugin is available
vad_meta = manager.get_discovered_meta("cjm-media-plugin-silero-vad")
if vad_meta:
    print(f"Found plugin: {vad_meta.name} v{vad_meta.version}")
else:
    print("Silero VAD plugin not found - install via plugins.yaml")

[PluginManager] Discovered manifest: cjm-media-plugin-silero-vad from /mnt/SN850X_8TB_EXT4/Projects/GitHub/cj-mills/cjm-transcript-vad-align/.cjm/manifests/cjm-media-plugin-silero-vad.json

Discovered 1 plugins from /mnt/SN850X_8TB_EXT4/Projects/GitHub/cj-mills/cjm-transcript-vad-align/.cjm/manifests
Found plugin: cjm-media-plugin-silero-vad v0.0.2

# Initialize and test AlignmentService
if vad_meta:
    # Load the plugin
    manager.load_plugin(vad_meta, {"threshold": 0.5})
    
    align_service = AlignmentService(manager)
    print(f"Plugin available: {align_service.is_available()}")
    
    # Test audio analysis (use await directly - Jupyter supports top-level await)
    audio_file = "/mnt/SN850X_8TB_EXT4/Projects/GitHub/cj-mills/cjm-transcription-plugin-voxtral-hf/test_files/02 - 1. Laying Plans.mp3"
    
    chunks, duration = await align_service.analyze_audio_async(audio_file)
    
    print(f"\nAudio duration: {duration:.2f}s")
    print(f"Detected {len(chunks)} VAD chunks")
    print("\nFirst 5 chunks:")
    for chunk in chunks[:5]:
        print(f"  [{chunk.index}] {chunk.start_time:.2f}s - {chunk.end_time:.2f}s (duration: {chunk.duration:.2f}s)")

[PluginManager] Launching worker for cjm-media-plugin-silero-vad...

[cjm-media-plugin-silero-vad] Starting worker on port 52517...
[cjm-media-plugin-silero-vad] Logs: /home/innom-dt/.cjm/logs/cjm-media-plugin-silero-vad.log

[PluginManager] HTTP Request: GET http://127.0.0.1:52517/health "HTTP/1.1 200 OK"
[PluginManager] HTTP Request: POST http://127.0.0.1:52517/initialize "HTTP/1.1 200 OK"
[PluginManager] Loaded plugin: cjm-media-plugin-silero-vad
[PluginManager] HTTP Request: POST http://127.0.0.1:52517/execute "HTTP/1.1 200 OK"

[cjm-media-plugin-silero-vad] Worker ready.
Plugin available: True

Audio duration: 256.39s
Detected 89 VAD chunks

First 5 chunks:
  [0] 1.10s - 2.30s (duration: 1.20s)
  [1] 4.80s - 5.90s (duration: 1.10s)
  [2] 6.60s - 9.80s (duration: 3.20s)
  [3] 10.70s - 12.40s (duration: 1.70s)
  [4] 12.60s - 15.40s (duration: 2.80s)

# Cleanup
if vad_meta:
    manager.unload_all()
    print("Plugins unloaded")

[PluginManager] HTTP Request: POST http://127.0.0.1:52517/cleanup "HTTP/1.1 200 OK"
[PluginManager] Unloaded plugin: cjm-media-plugin-silero-vad

Plugins unloaded