Función de Grabación y Transcripción en Tiempo Real para AMD A4-9125
Objetivo
Crear una función integrada que permita:
- 🎙️ Grabar audio en vivo desde micrófono
- ⚡ Transcribir automáticamente usando whisper-amd
- 📝 Guardar resultados en formatos múltiples
- 🔄 Integrar con ProcessingAgent existente
Script Principal de Grabación y Transcripción
#!/usr/bin/env python3
# live_transcriber_amd.py
import pyaudio
import wave
import time
import subprocess
import sys
import threading
import queue
from pathlib import Path
from datetime import datetime
import os
class LiveTranscriberAMD:
"""
Grabador y transcritor en tiempo real optimizado para AMD A4-9125.
Usa whisper-amd nativo para máxima velocidad.
"""
def __init__(self):
self.setup_paths()
self.setup_audio()
self.setup_whisper()
# Estado de grabación
self.recording = False
self.frames = []
self.stream = None
self.p = None
# Cola para procesamiento en paralelo
self.transcription_queue = queue.Queue()
print("🎙️ LiveTranscriberAMD inicializado para AMD A4-9125")
print(f"📁 Grabaciones: {self.recordings_dir}")
print(f"📝 Transcripciones: {self.transcripts_dir}")
def setup_paths(self):
"""Configurar directorios de trabajo."""
self.base_dir = Path.cwd()
self.recordings_dir = self.base_dir / "recordings" / "live"
self.transcripts_dir = self.base_dir / "recordings" / "transcripts"
# Crear directorios si no existen
self.recordings_dir.mkdir(parents=True, exist_ok=True)
self.transcripts_dir.mkdir(parents=True, exist_ok=True)
def setup_audio(self):
"""Configurar parámetros de audio optimizados."""
# Configuración optimizada para AMD A4-9125
self.CHUNK = 1024
self.FORMAT = pyaudio.paInt16
self.CHANNELS = 1 # Mono para mejor rendimiento
self.RATE = 16000 # 16kHz óptimo para whisper
print(f"🔧 Audio: {self.RATE}Hz, {self.CHANNELS} canal, chunks de {self.CHUNK}")
def setup_whisper(self):
"""Configurar whisper-amd."""
self.whisper_binary = "/usr/local/bin/whisper-amd"
self.models_dir = Path.home() / "whisper_models"
# Modelos disponibles (en orden de velocidad)
self.models = {
"tiny": self.models_dir / "ggml-tiny.bin", # Más rápido
"base": self.models_dir / "ggml-base.bin", # Equilibrado
"small": self.models_dir / "ggml-small.bin" # Más preciso
}
# Verificar whisper-amd
if not Path(self.whisper_binary).exists():
print(f"❌ ERROR: whisper-amd no encontrado en {self.whisper_binary}")
sys.exit(1)
print(f"✅ whisper-amd encontrado: {self.whisper_binary}")
# Verificar modelos
available_models = []
for name, path in self.models.items():
if path.exists():
size_mb = path.stat().st_size // (1024 * 1024)
available_models.append(f"{name} ({size_mb}MB)")
print(f"✅ Modelo {name}: {path.name}")
else:
print(f"❌ Modelo {name} no encontrado: {path}")
if not available_models:
print("❌ ERROR: No hay modelos disponibles")
sys.exit(1)
print(f"🧠 Modelos disponibles: {', '.join(available_models)}")
def get_audio_devices(self):
"""Listar dispositivos de audio disponibles."""
p = pyaudio.PyAudio()
print("\n🎤 DISPOSITIVOS DE AUDIO DISPONIBLES:")
print("=" * 50)
input_devices = []
for i in range(p.get_device_count()):
info = p.get_device_info_by_index(i)
if info['maxInputChannels'] > 0:
input_devices.append((i, info['name']))
print(f" [{i}] {info['name']}")
print(f" Canales: {info['maxInputChannels']}")
print(f" Frecuencia: {info['defaultSampleRate']}Hz")
print()
p.terminate()
return input_devices
def select_input_device(self):
"""Seleccionar dispositivo de entrada."""
devices = self.get_audio_devices()
if not devices:
print("❌ No se encontraron dispositivos de entrada")
return None
print("💡 Selecciona un dispositivo de entrada:")
try:
choice = int(input("Ingresa el número del dispositivo [0]: ") or "0")
if 0 <= choice < len(devices):
device_index = devices[choice][0]
device_name = devices[choice][1]
print(f"✅ Seleccionado: [{device_index}] {device_name}")
return device_index
else:
print("❌ Selección inválida, usando dispositivo por defecto")
return None
except ValueError:
print("❌ Entrada inválida, usando dispositivo por defecto")
return None
def start_recording(self, device_index=None):
"""Iniciar grabación de audio."""
if self.recording:
print("⚠️ Ya se está grabando")
return False
try:
self.p = pyaudio.PyAudio()
# Configurar stream de audio
stream_params = {
'format': self.FORMAT,
'channels': self.CHANNELS,
'rate': self.RATE,
'input': True,
'frames_per_buffer': self.CHUNK
}
if device_index is not None:
stream_params['input_device_index'] = device_index
self.stream = self.p.open(**stream_params)
self.frames = []
self.recording = True
self.start_time = time.time()
print("🔴 GRABACIÓN INICIADA")
print("💡 Presiona Ctrl+C para detener")
return True
except Exception as e:
print(f"❌ Error iniciando grabación: {e}")
return False
def record_chunk(self):
"""Grabar un chunk de audio."""
if not self.recording or not self.stream:
return None
try:
data = self.stream.read(self.CHUNK, exception_on_overflow=False)
self.frames.append(data)
return data
except Exception as e:
print(f"❌ Error grabando chunk: {e}")
return None
def stop_recording(self):
"""Detener grabación y guardar archivo."""
if not self.recording:
print("⚠️ No se está grabando")
return None
try:
self.stream.stop_stream()
self.stream.close()
self.p.terminate()
self.recording = False
# Calcular duración
end_time = time.time()
duration = end_time - self.start_time
# Generar nombre de archivo único
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"live_recording_{timestamp}.wav"
filepath = self.recordings_dir / filename
# Guardar archivo WAV
with wave.open(str(filepath), 'wb') as wf:
wf.setnchannels(self.CHANNELS)
wf.setsampwidth(self.p.get_sample_size(self.FORMAT))
wf.setframerate(self.RATE)
wf.writeframes(b''.join(self.frames))
file_size = filepath.stat().st_size / 1024 # KB
print(f"🟢 GRABACIÓN DETENIDA")
print(f"📁 Archivo: {filename}")
print(f"⏱️ Duración: {duration:.1f}s")
print(f"💾 Tamaño: {file_size:.1f}KB")
return {
'filepath': filepath,
'duration': duration,
'size_kb': file_size,
'timestamp': timestamp
}
except Exception as e:
print(f"❌ Error deteniendo grabación: {e}")
return None
def transcribe_file(self, audio_file, model="base", language="es"):
"""Transcribir archivo de audio usando whisper-amd."""
model_path = self.models.get(model)
if not model_path or not model_path.exists():
print(f"❌ Modelo {model} no disponible")
return None
print(f"\n🧠 INICIANDO TRANSCRIPCIÓN")
print(f"📁 Archivo: {audio_file.name}")
print(f"🤖 Modelo: {model}")
print(f"🗣️ Idioma: {language}")
# Comando whisper-amd optimizado para A4-9125
cmd = [
self.whisper_binary,
"-m", str(model_path),
"-t", "2", # 2 threads para A4-9125
"-p", "1", # 1 processor
"-l", language,
"--output-txt",
"--output-srt",
"--output-vtt",
"--output-dir", str(self.transcripts_dir),
str(audio_file)
]
try:
start_time = time.time()
# Ejecutar transcripción
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=300 # 5 minutos máximo
)
end_time = time.time()
transcription_time = end_time - start_time
if result.returncode == 0:
# Buscar archivo de texto generado
base_name = audio_file.stem
txt_file = self.transcripts_dir / f"{base_name}.txt"
srt_file = self.transcripts_dir / f"{base_name}.srt"
if txt_file.exists():
with open(txt_file, 'r', encoding='utf-8') as f:
transcript_text = f.read().strip()
word_count = len(transcript_text.split())
print(f"✅ TRANSCRIPCIÓN COMPLETADA")
print(f"⏱️ Tiempo: {transcription_time:.1f}s")
print(f"📊 Palabras: {word_count}")
print(f"📝 Texto: {txt_file.name}")
if srt_file.exists():
print(f"🎬 Subtítulos: {srt_file.name}")
return {
'text': transcript_text,
'txt_file': txt_file,
'srt_file': srt_file if srt_file.exists() else None,
'word_count': word_count,
'transcription_time': transcription_time,
'model': model,
'language': language
}
else:
print(f"❌ Archivo de transcripción no encontrado")
else:
print(f"❌ Error en transcripción:")
print(result.stderr)
except subprocess.TimeoutExpired:
print(f"❌ Timeout en transcripción (5 minutos)")
except Exception as e:
print(f"❌ Error durante transcripción: {e}")
return None
def record_and_transcribe(self, duration=None, model="base", language="es", device_index=None):
"""Grabar y transcribir en una sola función."""
print(f"\n🚀 GRABACIÓN Y TRANSCRIPCIÓN EN VIVO")
print(f"🧠 Modelo: {model}")
print(f"🗣️ Idioma: {language}")
if duration:
print(f"⏱️ Duración: {duration}s")
else:
print(f"⏱️ Duración: Hasta Ctrl+C")
print("=" * 50)
# Iniciar grabación
if not self.start_recording(device_index):
return None
try:
if duration:
# Grabación por tiempo fijo
chunks_to_record = int(self.RATE / self.CHUNK * duration)
for i in range(chunks_to_record):
self.record_chunk()
# Mostrar progreso
progress = (i + 1) / chunks_to_record * 100
print(f"\r🔴 Grabando... {progress:.1f}%", end='', flush=True)
print() # Nueva línea
else:
# Grabación hasta Ctrl+C
while True:
self.record_chunk()
time.sleep(0.01) # Pequeña pausa
except KeyboardInterrupt:
print(f"\n🛑 Grabación interrumpida por usuario")
# Detener y guardar
recording_result = self.stop_recording()
if not recording_result:
return None
# Transcribir automáticamente
print(f"\n⚡ Iniciando transcripción automática...")
transcription_result = self.transcribe_file(
recording_result['filepath'],
model=model,
language=language
)
if transcription_result:
print(f"\n📋 RESULTADO FINAL:")
print(f"🎙️ Audio: {recording_result['filepath'].name}")
print(f"📝 Transcripción: {transcription_result['txt_file'].name}")
print(f"⏱️ Audio: {recording_result['duration']:.1f}s")
print(f"⚡ Transcripción: {transcription_result['transcription_time']:.1f}s")
print(f"📊 Palabras: {transcription_result['word_count']}")
# Mostrar preview del texto
preview = transcription_result['text'][:200]
print(f"\n📖 PREVIEW:")
print(f"'{preview}{'...' if len(transcription_result['text']) > 200 else ''}'")
return {
'recording': recording_result,
'transcription': transcription_result
}
return {'recording': recording_result, 'transcription': None}
def interactive_mode(self):
"""Modo interactivo para grabación y transcripción."""
print(f"\n🎙️ MODO INTERACTIVO - LiveTranscriberAMD")
print("=" * 50)
while True:
print(f"\n💡 OPCIONES:")
print("1. 🎙️ Grabar y transcribir (duración fija)")
print("2. 🔴 Grabar y transcribir (hasta Ctrl+C)")
print("3. 🎤 Seleccionar dispositivo de audio")
print("4. 📁 Ver archivos recientes")
print("5. ❌ Salir")
try:
choice = input("\nSelecciona una opción [1-5]: ").strip()
if choice == "1":
# Grabación con duración fija
try:
duration = int(input("Duración en segundos [30]: ") or "30")
model = input("Modelo [base]: ").strip() or "base"
language = input("Idioma [es]: ").strip() or "es"
device_index = getattr(self, 'selected_device', None)
self.record_and_transcribe(duration, model, language, device_index)
except ValueError:
print("❌ Duración inválida")
elif choice == "2":
# Grabación hasta Ctrl+C
model = input("Modelo [base]: ").strip() or "base"
language = input("Idioma [es]: ").strip() or "es"
device_index = getattr(self, 'selected_device', None)
self.record_and_transcribe(None, model, language, device_index)
elif choice == "3":
# Seleccionar dispositivo
self.selected_device = self.select_input_device()
elif choice == "4":
# Ver archivos recientes
self.show_recent_files()
elif choice == "5":
print("👋 ¡Hasta luego!")
break
else:
print("❌ Opción inválida")
except KeyboardInterrupt:
print(f"\n👋 ¡Hasta luego!")
break
except Exception as e:
print(f"❌ Error: {e}")
def show_recent_files(self):
"""Mostrar archivos recientes."""
print(f"\n📁 ARCHIVOS RECIENTES:")
print("=" * 50)
# Archivos de audio
audio_files = sorted(
self.recordings_dir.glob("*.wav"),
key=lambda x: x.stat().st_mtime,
reverse=True
)[:5]
print("🎙️ Grabaciones:")
for f in audio_files:
size_kb = f.stat().st_size / 1024
mtime = datetime.fromtimestamp(f.stat().st_mtime)
print(f" 📁 {f.name} ({size_kb:.1f}KB) - {mtime.strftime('%Y-%m-%d %H:%M')}")
# Archivos de transcripción
txt_files = sorted(
self.transcripts_dir.glob("*.txt"),
key=lambda x: x.stat().st_mtime,
reverse=True
)[:5]
print("\n📝 Transcripciones:")
for f in txt_files:
mtime = datetime.fromtimestamp(f.stat().st_mtime)
print(f" 📄 {f.name} - {mtime.strftime('%Y-%m-%d %H:%M')}")
def main():
"""Función principal."""
print("🚀 Iniciando LiveTranscriberAMD para AMD A4-9125")
try:
transcriber = LiveTranscriberAMD()
# Verificar argumentos de línea de comandos
if len(sys.argv) > 1:
if sys.argv[1] == "--quick":
# Modo rápido: grabar 30 segundos
print("⚡ Modo rápido: grabando 30 segundos...")
transcriber.record_and_transcribe(duration=30)
elif sys.argv[1] == "--live":
# Modo en vivo: hasta Ctrl+C
print("🔴 Modo en vivo: presiona Ctrl+C para detener...")
transcriber.record_and_transcribe()
else:
print("❌ Argumento desconocido. Usa --quick o --live")
else:
# Modo interactivo
transcriber.interactive_mode()
except KeyboardInterrupt:
print(f"\n👋 Programa terminado por usuario")
except Exception as e:
print(f"❌ Error fatal: {e}")
if __name__ == "__main__":
main()Script de Instalación y Configuración
#!/bin/bash
# setup_live_transcriber.sh
echo "🔧 CONFIGURANDO LiveTranscriberAMD"
echo "================================="
# Verificar dependencias Python
echo "📦 Verificando dependencias Python..."
python3 -c "import pyaudio, wave, subprocess, threading, queue" 2>/dev/null || {
echo "❌ Faltan dependencias Python"
echo "💡 Instalando pyaudio..."
sudo apt update
sudo apt install -y python3-pyaudio portaudio19-dev
pip3 install pyaudio
}
# Verificar whisper-amd
echo "🤖 Verificando whisper-amd..."
if ! command -v whisper-amd >/dev/null 2>&1; then
echo "❌ whisper-amd no encontrado"
echo "💡 Asegúrate de haber completado los pasos anteriores"
exit 1
fi
echo "✅ whisper-amd encontrado"
# Verificar modelos
echo "🧠 Verificando modelos..."
MODELS_DIR="$HOME/whisper_models"
if [ ! -d "$MODELS_DIR" ]; then
echo "❌ Directorio de modelos no encontrado: $MODELS_DIR"
exit 1
fi
model_count=0
for model in tiny base small; do
if [ -f "$MODELS_DIR/ggml-$model.bin" ]; then
size=$(stat -c%s "$MODELS_DIR/ggml-$model.bin" 2>/dev/null | awk '{print int($1/1024/1024)"MB"}')
echo "✅ Modelo $model: $size"
((model_count++))
else
echo "⚠️ Modelo $model no encontrado"
fi
done
if [ $model_count -eq 0 ]; then
echo "❌ No hay modelos disponibles"
exit 1
fi
# Crear directorios
echo "📁 Creando directorios..."
mkdir -p recordings/live
mkdir -p recordings/transcripts
echo "✅ Directorios creados"
# Hacer ejecutable el script
echo "🔧 Configurando permisos..."
chmod +x live_transcriber_amd.py
# Crear alias útiles
echo "⚡ Creando aliases..."
cat >> ~/.bashrc << 'EOF'
# LiveTranscriberAMD aliases
alias transcribe-quick="python3 live_transcriber_amd.py --quick"
alias transcribe-live="python3 live_transcriber_amd.py --live"
alias transcribe-interactive="python3 live_transcriber_amd.py"
EOF
echo "✅ Configuración completada"
echo ""
echo "🚀 COMANDOS DISPONIBLES:"
echo "========================"
echo "python3 live_transcriber_amd.py # Modo interactivo"
echo "python3 live_transcriber_amd.py --quick # Grabar 30s"
echo "python3 live_transcriber_amd.py --live # Hasta Ctrl+C"
echo ""
echo "💡 Recarga tu terminal o ejecuta: source ~/.bashrc"Integración con ProcessingAgent
# agents/processing_agent_live.py
import sys
from pathlib import Path
# Importar el LiveTranscriberAMD
sys.path.append(str(Path(__file__).parent.parent))
from live_transcriber_amd import LiveTranscriberAMD
# Importar ProcessingAgent original
from agents.processing_agent import ProcessingAgent
class ProcessingAgentLive(ProcessingAgent):
"""
ProcessingAgent extendido con capacidades de grabación en vivo.
"""
def __init__(self):
super().__init__()
self.live_transcriber = LiveTranscriberAMD()
print(f"[{self.agent_name}] Extendido con capacidades de grabación en vivo")
def record_and_transcribe_live(self, duration=None, model="base", language="es"):
"""
Grabar y transcribir en una sola operación.
Se integra con el flujo de ProcessingAgent.
"""
print(f"[{self.agent_name}] Iniciando grabación y transcripción en vivo...")
# Usar LiveTranscriberAMD para grabar y transcribir
result = self.live_transcriber.record_and_transcribe(
duration=duration,
model=model,
language=language
)
if result and result.get('transcription'):
# Integrar con el flujo del ProcessingAgent
transcription = result['transcription']
# Crear resultado compatible con ProcessingAgent
return {
"text": transcription['text'],
"path": str(transcription['txt_file']),
"language": transcription['language'],
"method": "live-whisper-amd",
"model": transcription['model'],
"duration_seconds": result['recording']['duration'],
"transcription_time": transcription['transcription_time'],
"word_count": transcription['word_count']
}
return None
def run_live_demo(self):
"""Demo de grabación en vivo."""
print(f"[{self.agent_name}] Ejecutando demo de grabación en vivo...")
# Grabar 15 segundos como demo
result = self.record_and_transcribe_live(duration=15, model="tiny", language="es")
if result:
print(f"[{self.agent_name}] Demo completado exitosamente:")
print(f" Texto: {result['text'][:100]}...")
print(f" Archivo: {result['path']}")
print(f" Duración audio: {result['duration_seconds']:.1f}s")
print(f" Tiempo transcripción: {result['transcription_time']:.1f}s")
print(f" Palabras: {result['word_count']}")
else:
print(f"[{self.agent_name}] Demo falló")
if __name__ == "__main__":
agent = ProcessingAgentLive()
agent.run_live_demo()
Función de Grabación y Transcripción en Tiempo Real para AMD A4-9125
Objetivo
Crear una función integrada que permita:
Script Principal de Grabación y Transcripción
Script de Instalación y Configuración
Integración con ProcessingAgent