mirror of
https://github.com/QuentinFuxa/WhisperLiveKit.git
synced 2026-03-07 22:33:36 +00:00
Logs for https://github.com/QuentinFuxa/WhisperLiveKit/issues/110 https://github.com/QuentinFuxa/WhisperLiveKit/issues/106
https://github.com/QuentinFuxa/WhisperLiveKit/issues/90 https://github.com/QuentinFuxa/WhisperLiveKit/issues/87 https://github.com/QuentinFuxa/WhisperLiveKit/issues/81 https://github.com/QuentinFuxa/WhisperLiveKit/issues/2
This commit is contained in:
Quentin Fuxa
@@ -6,7 +6,6 @@ import math
|
||||
import logging
|
||||
import traceback
|
||||
from datetime import timedelta
|
||||
from typing import List, Dict, Any
|
||||
from whisperlivekit.timed_objects import ASRToken
|
||||
from whisperlivekit.whisper_streaming_custom.whisper_online import online_factory
|
||||
from whisperlivekit.core import WhisperLiveKit
|
||||
@@ -39,7 +38,10 @@ class AudioProcessor:
|
||||
self.bytes_per_sample = 2
|
||||
self.bytes_per_sec = self.samples_per_sec * self.bytes_per_sample
|
||||
self.max_bytes_per_sec = 32000 * 5 # 5 seconds of audio at 32 kHz
|
||||
|
||||
self.last_ffmpeg_activity = time()
|
||||
self.ffmpeg_health_check_interval = 5
|
||||
self.ffmpeg_max_idle_time = 10
|
||||
|
||||
# State management
|
||||
self.tokens = []
|
||||
self.buffer_transcription = ""
|
||||
@@ -78,14 +80,50 @@ class AudioProcessor:
|
||||
|
||||
async def restart_ffmpeg(self):
|
||||
"""Restart the FFmpeg process after failure."""
|
||||
logger.warning("Restarting FFmpeg process...")
|
||||
|
||||
if self.ffmpeg_process:
|
||||
try:
|
||||
self.ffmpeg_process.kill()
|
||||
await asyncio.get_event_loop().run_in_executor(None, self.ffmpeg_process.wait)
|
||||
# we check if process is still running
|
||||
if self.ffmpeg_process.poll() is None:
|
||||
logger.info("Terminating existing FFmpeg process")
|
||||
self.ffmpeg_process.stdin.close()
|
||||
self.ffmpeg_process.terminate()
|
||||
|
||||
# wait for termination with timeout
|
||||
try:
|
||||
await asyncio.wait_for(
|
||||
asyncio.get_event_loop().run_in_executor(None, self.ffmpeg_process.wait),
|
||||
timeout=5.0
|
||||
)
|
||||
except asyncio.TimeoutError:
|
||||
logger.warning("FFmpeg process did not terminate, killing forcefully")
|
||||
self.ffmpeg_process.kill()
|
||||
await asyncio.get_event_loop().run_in_executor(None, self.ffmpeg_process.wait)
|
||||
except Exception as e:
|
||||
logger.warning(f"Error killing FFmpeg process: {e}")
|
||||
logger.error(f"Error during FFmpeg process termination: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
|
||||
# we start new process
|
||||
try:
|
||||
logger.info("Starting new FFmpeg process")
|
||||
self.ffmpeg_process = self.start_ffmpeg_decoder()
|
||||
self.pcm_buffer = bytearray()
|
||||
self.last_ffmpeg_activity = time()
|
||||
logger.info("FFmpeg process restarted successfully")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to restart FFmpeg process: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
# try again after 5s
|
||||
await asyncio.sleep(5)
|
||||
try:
|
||||
self.ffmpeg_process = self.start_ffmpeg_decoder()
|
||||
self.pcm_buffer = bytearray()
|
||||
self.last_ffmpeg_activity = time()
|
||||
logger.info("FFmpeg process restarted successfully on second attempt")
|
||||
except Exception as e2:
|
||||
logger.critical(f"Failed to restart FFmpeg process on second attempt: {e2}")
|
||||
logger.critical(traceback.format_exc())
|
||||
|
||||
async def update_transcription(self, new_tokens, buffer, end_buffer, full_transcription, sep):
|
||||
"""Thread-safe update of transcription with new data."""
|
||||
@@ -154,21 +192,33 @@ class AudioProcessor:
|
||||
|
||||
while True:
|
||||
try:
|
||||
# Calculate buffer size based on elapsed time
|
||||
elapsed_time = math.floor((time() - beg) * 10) / 10 # Round to 0.1 sec
|
||||
current_time = time()
|
||||
elapsed_time = math.floor((current_time - beg) * 10) / 10
|
||||
buffer_size = max(int(32000 * elapsed_time), 4096)
|
||||
beg = time()
|
||||
beg = current_time
|
||||
|
||||
# Read chunk with timeout
|
||||
# Detect idle state much more quickly
|
||||
if current_time - self.last_ffmpeg_activity > self.ffmpeg_max_idle_time:
|
||||
logger.warning(f"FFmpeg process idle for {current_time - self.last_ffmpeg_activity:.2f}s. Restarting...")
|
||||
await self.restart_ffmpeg()
|
||||
beg = time()
|
||||
self.last_ffmpeg_activity = time()
|
||||
continue
|
||||
|
||||
# Reduce timeout for reading from FFmpeg
|
||||
try:
|
||||
chunk = await asyncio.wait_for(
|
||||
loop.run_in_executor(None, self.ffmpeg_process.stdout.read, buffer_size),
|
||||
timeout=15.0
|
||||
timeout=5.0 # Shorter timeout (5 seconds instead of 15)
|
||||
)
|
||||
if chunk:
|
||||
self.last_ffmpeg_activity = time()
|
||||
|
||||
except asyncio.TimeoutError:
|
||||
logger.warning("FFmpeg read timeout. Restarting...")
|
||||
await self.restart_ffmpeg()
|
||||
beg = time()
|
||||
self.last_ffmpeg_activity = time()
|
||||
continue
|
||||
|
||||
if not chunk:
|
||||
@@ -366,7 +416,7 @@ class AudioProcessor:
|
||||
logger.warning(f"Exception in results_formatter: {e}")
|
||||
logger.warning(f"Traceback: {traceback.format_exc()}")
|
||||
await asyncio.sleep(0.5) # Back off on error
|
||||
|
||||
|
||||
async def create_tasks(self):
|
||||
"""Create and start processing tasks."""
|
||||
|
||||
@@ -378,6 +428,35 @@ class AudioProcessor:
|
||||
tasks.append(asyncio.create_task(self.diarization_processor(self.diarization)))
|
||||
|
||||
tasks.append(asyncio.create_task(self.ffmpeg_stdout_reader()))
|
||||
|
||||
# Monitor overall system health
|
||||
async def watchdog():
|
||||
while True:
|
||||
try:
|
||||
await asyncio.sleep(10) # Check every 10 seconds instead of 60
|
||||
|
||||
current_time = time()
|
||||
# Check for stalled tasks
|
||||
for i, task in enumerate(tasks):
|
||||
if task.done():
|
||||
exc = task.exception() if task.done() else None
|
||||
task_name = task.get_name() if hasattr(task, 'get_name') else f"Task {i}"
|
||||
logger.error(f"{task_name} unexpectedly completed with exception: {exc}")
|
||||
|
||||
# Check for FFmpeg process health with shorter thresholds
|
||||
ffmpeg_idle_time = current_time - self.last_ffmpeg_activity
|
||||
if ffmpeg_idle_time > 15: # 15 seconds instead of 180
|
||||
logger.warning(f"FFmpeg idle for {ffmpeg_idle_time:.2f}s - may need attention")
|
||||
|
||||
# Force restart after 30 seconds of inactivity (instead of 600)
|
||||
if ffmpeg_idle_time > 30:
|
||||
logger.error("FFmpeg idle for too long, forcing restart")
|
||||
await self.restart_ffmpeg()
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error in watchdog task: {e}")
|
||||
|
||||
tasks.append(asyncio.create_task(watchdog()))
|
||||
self.tasks = tasks
|
||||
|
||||
return self.results_formatter()
|
||||
@@ -399,11 +478,34 @@ class AudioProcessor:
|
||||
|
||||
async def process_audio(self, message):
|
||||
"""Process incoming audio data."""
|
||||
try:
|
||||
self.ffmpeg_process.stdin.write(message)
|
||||
self.ffmpeg_process.stdin.flush()
|
||||
except (BrokenPipeError, AttributeError) as e:
|
||||
logger.warning(f"Error writing to FFmpeg: {e}. Restarting...")
|
||||
await self.restart_ffmpeg()
|
||||
self.ffmpeg_process.stdin.write(message)
|
||||
self.ffmpeg_process.stdin.flush()
|
||||
retry_count = 0
|
||||
max_retries = 3
|
||||
|
||||
# Log periodic heartbeats showing ongoing audio proc
|
||||
current_time = time()
|
||||
if not hasattr(self, '_last_heartbeat') or current_time - self._last_heartbeat >= 10:
|
||||
logger.debug(f"Processing audio chunk, last FFmpeg activity: {current_time - self.last_ffmpeg_activity:.2f}s ago")
|
||||
self._last_heartbeat = current_time
|
||||
|
||||
while retry_count < max_retries:
|
||||
try:
|
||||
if not self.ffmpeg_process or not hasattr(self.ffmpeg_process, 'stdin') or self.ffmpeg_process.poll() is not None:
|
||||
logger.warning("FFmpeg process not available, restarting...")
|
||||
await self.restart_ffmpeg()
|
||||
|
||||
self.ffmpeg_process.stdin.write(message)
|
||||
self.ffmpeg_process.stdin.flush()
|
||||
self.last_ffmpeg_activity = time() # Update activity timestamp
|
||||
return
|
||||
|
||||
except (BrokenPipeError, AttributeError, OSError) as e:
|
||||
retry_count += 1
|
||||
logger.warning(f"Error writing to FFmpeg: {e}. Retry {retry_count}/{max_retries}...")
|
||||
|
||||
if retry_count < max_retries:
|
||||
await self.restart_ffmpeg()
|
||||
await asyncio.sleep(0.5) # Shorter pause between retries
|
||||
else:
|
||||
logger.error("Maximum retries reached for FFmpeg process")
|
||||
await self.restart_ffmpeg()
|
||||
return
|
||||
Reference in New Issue
Block a user