#!/usr/bin/env python3 import argparse import json import subprocess import sys import time import numpy as np import torch def parse_track_pairs(pairs_list): """Parses track pairs formatted as '1 sacha 2 tutor' or 'device1=sacha device2=tutor'.""" track_map = {} if not pairs_list: return track_map if len(pairs_list) == 1 and " " in pairs_list[0]: pairs_list = pairs_list[0].split() i = 0 while i < len(pairs_list): item = pairs_list[i] if "=" in item: key, label = item.split("=", 1) track_map[key] = label i += 1 else: if i + 1 < len(pairs_list): key = item label = pairs_list[i + 1] track_map[key] = label i += 2 else: raise ValueError(f"Unmatched key without label: {item}") return track_map def load_silero_vad(): """Loads Silero VAD model via PyTorch Hub.""" model, _ = torch.hub.load( repo_or_dir='snakers4/silero-vad', model='silero_vad', force_reload=False, onnx=False ) return model def process_file_track_streamed(file_path: str, track_id: int, label: str, model, min_silence_ms: int): """Streams audio from FFmpeg in 32ms frames directly into Silero VAD.""" target_sr = 16000 frame_samples = 512 frame_bytes = frame_samples * 2 # s16le = 2 bytes min_silence_frames = int((min_silence_ms / 1000.0) / 0.032) threshold = 0.5 stream_index = track_id - 1 cmd = [ "ffmpeg", "-loglevel", "error", "-threads", "0", "-i", file_path, "-map", f"0:a:{stream_index}", "-ac", "1", "-ar", str(target_sr), "-f", "s16le", "-" ] try: process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) except Exception as e: sys.stderr.write(f"Error running FFmpeg on track {track_id}: {e}\n") return [] model.reset_states() segments = [] is_speaking = False speech_start_sec = 0.0 silence_frame_count = 0 current_sample_idx = 0 with torch.no_grad(): while True: raw_bytes = process.stdout.read(frame_bytes) if len(raw_bytes) < frame_bytes: if is_speaking: end_sec = current_sample_idx / target_sr segments.append({"label": label, "start": round(speech_start_sec, 3), "end": round(end_sec, 3)}) break audio_int16 = np.frombuffer(raw_bytes, dtype=np.int16) tensor_frame = torch.from_numpy(audio_int16.astype(np.float32) / 32768.0) speech_prob = model(tensor_frame, target_sr).item() current_time_sec = current_sample_idx / target_sr if speech_prob >= threshold: silence_frame_count = 0 if not is_speaking: is_speaking = True speech_start_sec = current_time_sec else: if is_speaking: silence_frame_count += 1 if silence_frame_count >= min_silence_frames: is_speaking = False speech_end_sec = current_time_sec - (silence_frame_count * 0.032) segments.append({ "label": label, "start": round(speech_start_sec, 3), "end": round(speech_end_sec, 3) }) silence_frame_count = 0 current_sample_idx += frame_samples process.wait() return segments import copy import json import subprocess import sys import threading import time import numpy as np import torch class RealtimeVADStreamer: """Streams live audio directly from PulseAudio/PipeWire sources using parec.""" def __init__(self, device_name: str, label: str, model, min_silence_ms: int, out_file): self.device_name = device_name self.label = label # Deepcopy model to ensure thread safety self.model = copy.deepcopy(model) self.out_file = out_file self.sample_rate = 16000 # 512 samples @ 16kHz = 32ms frames (s16le = 2 bytes/sample = 1024 bytes) self.frame_samples = 512 self.frame_bytes = self.frame_samples * 2 self.min_silence_samples = int(self.sample_rate * (min_silence_ms / 1000.0)) self.is_speaking = False self.speech_start_sample = 0 self.silence_counter_samples = 0 self.running = False self.thread = None self.process = None def emit_event(self, event_type: str, timestamp: float, duration: float = None): payload = { "label": self.label, "event": event_type, "timestamp": round(timestamp, 3) } if duration is not None: payload["duration"] = round(duration, 3) self.out_file.write(json.dumps(payload) + "\n") self.out_file.flush() def _read_loop(self, start_epoch: float): cmd = [ "parec", "--device=" + self.device_name, "--format=s16le", "--rate=" + str(self.sample_rate), "--channels=1" ] try: self.process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) except Exception as e: sys.stderr.write(f"[{self.label}] Failed to start parec for device '{self.device_name}': {e}\n") return if hasattr(self.model, 'reset_states'): self.model.reset_states() total_samples_processed = 0 while self.running: raw_bytes = self.process.stdout.read(self.frame_bytes) if len(raw_bytes) < self.frame_bytes: break audio_int16 = np.frombuffer(raw_bytes, dtype=np.int16) tensor_frame = torch.from_numpy(audio_int16.astype(np.float32) / 32768.0) with torch.no_grad(): speech_prob = self.model(tensor_frame, self.sample_rate).item() threshold = 0.5 if speech_prob >= threshold: self.silence_counter_samples = 0 if not self.is_speaking: self.is_speaking = True self.speech_start_sample = total_samples_processed start_sec = self.speech_start_sample / self.sample_rate self.emit_event("start", start_sec) else: if self.is_speaking: self.silence_counter_samples += self.frame_samples if self.silence_counter_samples >= self.min_silence_samples: self.is_speaking = False # Speech ended right before the silence limit started speech_end_sample = (total_samples_processed + self.frame_samples) - self.silence_counter_samples start_sec = self.speech_start_sample / self.sample_rate end_sec = max(speech_end_sample / self.sample_rate, start_sec) duration = end_sec - start_sec self.emit_event("end", end_sec, duration=duration) self.silence_counter_samples = 0 total_samples_processed += self.frame_samples def start(self, start_epoch: float): self.running = True self.thread = threading.Thread(target=self._read_loop, args=(start_epoch,), daemon=True) self.thread.start() def stop(self): self.running = False if self.process: self.process.terminate() self.process.wait() def list_pulse_devices(): """Displays available audio input devices.""" import sounddevice as sd print("\n--- Available Audio Input Devices ---") devices = sd.query_devices() for idx, dev in enumerate(devices): if dev['max_input_channels'] > 0: print(f"ID/Name: {dev['name']} (Index: {idx})") print("------------------------------------\n") def main(): parser = argparse.ArgumentParser(description="Silero VAD for Files or PulseAudio Live Streaming.") group = parser.add_mutually_exclusive_group(required=True) group.add_argument("-f", "--file", help="Path to input audio/video file (e.g. MKV)") group.add_argument("-p", "--pulse", action="store_true", help="Use PulseAudio live source stream mode") group.add_argument("--list-devices", action="store_true", help="List available audio devices and exit") parser.add_argument( "-t", "--tracks", nargs="+", help="Track/Device and label pairs. File mode: '-t 1 sacha 2 tutor'. Pulse mode: '-t VirtualMicSink=sacha mic=tutor'" ) parser.add_argument("-s", "--silence-limit", type=int, default=300, help="Min silence duration in ms (default: 300)") parser.add_argument("-d", "--duration", type=float, default=0, help="Live capture duration in seconds for Pulse mode (0 for continuous/Ctrl+C)") parser.add_argument("-o", "--output", help="Output path for JSONL (default: stdout)") args = parser.parse_args() if args.list_devices: list_pulse_devices() return model = load_silero_vad() out_file = open(args.output, "w", encoding="utf-8") if args.output else sys.stdout try: if args.file: track_map = parse_track_pairs(args.tracks) if args.tracks else {1: "speaker"} all_segments = [] for track_id_str, label in track_map.items(): segments = process_file_track_streamed( file_path=args.file, track_id=int(track_id_str), label=label, model=model, min_silence_ms=args.silence_limit ) all_segments.extend(segments) all_segments.sort(key=lambda x: x["start"]) for segment in all_segments: out_file.write(json.dumps(segment) + "\n") elif args.pulse: if not args.tracks: sys.stderr.write("Error: PulseAudio mode requires source-label pairs using -t / --tracks\n") sys.exit(1) device_map = parse_track_pairs(args.tracks) streamers = [] for device_name, label in device_map.items(): streamer = RealtimeVADStreamer( device_name=device_name, label=label, model=model, min_silence_ms=args.silence_limit, out_file=out_file ) streamers.append(streamer) start_epoch = time.time() sys.stderr.write("Starting live PulseAudio streaming...\n") for s in streamers: s.start(start_epoch) try: if args.duration > 0: time.sleep(args.duration) else: while True: time.sleep(0.1) except KeyboardInterrupt: sys.stderr.write("\nStopping live capture...\n") finally: for s in streamers: s.stop() finally: if args.output: out_file.close() if __name__ == "__main__": main()