audio-capture-guide

0   alsa_input.pci-0000_00_1f.3.analog-stereo    module-alsa-card.c    s16le 2ch 44100Hz
1   alsa_input.usb-Blue_Microphones_Yeti_Stereo   module-alsa-card.c    s16le 2ch 48000Hz

import pyaudio
import numpy as np

audio = pyaudio.PyAudio()

# List devices
for i in range(audio.get_device_count()):
    info = audio.get_device_info_by_index(i)
    if info['maxInputChannels'] > 0:
        print(f"{i}: {info['name']} (inputs: {info['maxInputChannels']})")

# Open microphone stream
stream = audio.open(
    format=pyaudio.paInt16,
    channels=1,
    rate=16000,
    input=True,
    input_device_index=None,  # None = default, or specify device index
    frames_per_buffer=1024
)

# Read audio chunks
while True:
    audio_chunk = stream.read(1024)
    audio_np = np.frombuffer(audio_chunk, dtype=np.int16)
    # Process audio_np...

2   alsa_output.pci-0000_00_1f.3.analog-stereo.monitor   module-alsa-card.c   s16le 2ch 44100Hz

import pyaudio
import numpy as np
import threading

audio = pyaudio.PyAudio()

# Buffer to store combined audio
combined_buffer = []
buffer_lock = threading.Lock()

def capture_mic(device_index):
    stream = audio.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=16000,
        input=True,
        input_device_index=device_index,
        frames_per_buffer=1024
    )
    while True:
        data = stream.read(1024)
        audio_np = np.frombuffer(data, dtype=np.int16)
        with buffer_lock:
            combined_buffer.append(('mic', audio_np))

def capture_speakers(device_index):
    stream = audio.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=16000,
        input=True,
        input_device_index=device_index,
        frames_per_buffer=1024
    )
    while True:
        data = stream.read(1024)
        audio_np = np.frombuffer(data, dtype=np.int16)
        with buffer_lock:
            combined_buffer.append(('speaker', audio_np))

# Start capture threads
mic_thread = threading.Thread(target=capture_mic, args=(mic_device_index,))
speaker_thread = threading.Thread(target=capture_speakers, args=(monitor_device_index,))

mic_thread.daemon = True
speaker_thread.daemon = True

mic_thread.start()
speaker_thread.start()

# Process combined buffer
while True:
    with buffer_lock:
        if combined_buffer:
            source, audio_chunk = combined_buffer.pop(0)
            # Process audio_chunk...

#!/usr/bin/env python3
"""
Real-time audio transcription with AMD GPU support
Captures both mic and speakers, transcribes with openai-whisper
"""

import whisper
import pyaudio
import numpy as np
import subprocess
import sys
from threading import Thread, Lock
from queue import Queue

# Load Whisper model (ROCm-accelerated on AMD GPUs)
print("Loading Whisper model...")
model = whisper.load_model("base", device="cuda")  # "cuda" works for ROCm too
print("Model loaded!")

# Audio configuration
RATE = 16000
CHUNK = RATE * 2  # 2 seconds of audio
FORMAT = pyaudio.paInt16
CHANNELS = 1

# Transcription queue
audio_queue = Queue()

def get_monitor_device(name_contains="monitor"):
    """Find monitor device for speaker capture"""
    p = pyaudio.PyAudio()
    for i in range(p.get_device_count()):
        info = p.get_device_info_by_index(i)
        if name_contains.lower() in info['name'].lower() and info['maxInputChannels'] > 0:
            print(f"Found monitor device: {info['name']}")
            return i
    return None

def get_mic_device():
    """Find default microphone"""
    p = pyaudio.PyAudio()
    return p.get_default_input_device_info()['index']

def capture_audio_pulseaudio():
    """
    Capture combined audio using PulseAudio virtual sink
    Assumes you've set up the 'combined' sink as shown above
    """
    cmd = [
        'parec',
        '--device=combined.monitor',
        '--format=s16le',
        '--rate=16000',
        '--channels=1'
    ]

    process = subprocess.Popen(cmd, stdout=subprocess.PIPE, bufsize=CHUNK*2)

    print("Capturing audio from combined source (mic + speakers)...")

    while True:
        audio_data = process.stdout.read(CHUNK * 2)  # 2 bytes per sample
        if not audio_data:
            break

        audio_np = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0
        audio_queue.put(audio_np)

def capture_audio_pyaudio(source='mic'):
    """
    Capture audio using PyAudio
    source: 'mic', 'speakers', or 'both' (not implemented yet)
    """
    audio = pyaudio.PyAudio()

    if source == 'mic':
        device_index = get_mic_device()
        print(f"Capturing from microphone (device {device_index})...")
    elif source == 'speakers':
        device_index = get_monitor_device()
        print(f"Capturing from speakers (device {device_index})...")
    else:
        print("Invalid source. Use 'mic' or 'speakers'")
        return

    stream = audio.open(
        format=FORMAT,
        channels=CHANNELS,
        rate=RATE,
        input=True,
        input_device_index=device_index,
        frames_per_buffer=CHUNK
    )

    while True:
        audio_data = stream.read(CHUNK, exception_on_overflow=False)
        audio_np = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0
        audio_queue.put(audio_np)

def transcribe_worker():
    """Process audio chunks and transcribe"""
    print("Transcription worker started...")

    while True:
        audio_chunk = audio_queue.get()

        if audio_chunk is None:
            break

        # Check if audio has speech (simple energy threshold)
        energy = np.abs(audio_chunk).mean()
        if energy < 0.01:  # Silence threshold
            continue

        # Transcribe
        try:
            result = model.transcribe(
                audio_chunk,
                fp16=False,  # Use fp32 for ROCm
                language='en',  # Specify if known
                task='transcribe'
            )

            text = result["text"].strip()
            if text:
                print(f"[TRANSCRIPT]: {text}")

        except Exception as e:
            print(f"Transcription error: {e}")

if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description='Real-time audio transcription')
    parser.add_argument('--source', choices=['mic', 'speakers', 'combined'],
                       default='combined', help='Audio source to capture')
    args = parser.parse_args()

    # Start transcription worker
    transcribe_thread = Thread(target=transcribe_worker, daemon=True)
    transcribe_thread.start()

    # Start audio capture
    try:
        if args.source == 'combined':
            capture_audio_pulseaudio()
        else:
            capture_audio_pyaudio(args.source)
    except KeyboardInterrupt:
        print("\nStopping...")
        audio_queue.put(None)
        sys.exit(0)

import os

# May need to override GPU architecture
os.environ["HSA_OVERRIDE_GFX_VERSION"] = "11.0.1"  # For RX 7800 XT (gfx1101)

# Load model
model = whisper.load_model("base", device="cuda")  # "cuda" works for ROCm

# Transcribe with fp32 (fp16 can be problematic on ROCm)
result = model.transcribe(audio, fp16=False)

# Smaller chunks = lower latency, less context
CHUNK = RATE * 1  # 1 second (faster, less accurate)

# Larger chunks = higher latency, better context
CHUNK = RATE * 3  # 3 seconds (slower, more accurate)

# Recommended for real-time
CHUNK = RATE * 2  # 2 seconds (good balance)

import webrtcvad

vad = webrtcvad.Vad(3)  # Aggressiveness 0-3

def has_speech(audio_chunk, sample_rate=16000):
    """Check if audio chunk contains speech"""
    # Convert to bytes
    audio_bytes = (audio_chunk * 32768).astype(np.int16).tobytes()

    # VAD expects 10, 20, or 30ms frames
    frame_duration = 30  # ms
    frame_size = int(sample_rate * frame_duration / 1000) * 2  # 2 bytes per sample

    # Check first frame
    return vad.is_speech(audio_bytes[:frame_size], sample_rate)

# In transcription loop
if has_speech(audio_chunk):
    result = model.transcribe(audio_chunk)