File size: 3,223 Bytes
91939bc |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 |
"""
Error handling utilities for VidGen
"""
import streamlit as st
import time
import functools
from typing import Any, Callable, Optional
def retry_on_error(max_retries: int = 3, delay: float = 1.0, backoff: float = 2.0):
"""Decorator to retry functions on specific errors"""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs) -> Any:
last_exception = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
error_msg = str(e).lower()
# Check for retryable errors
retryable_errors = [
'bodystreamBuffer was aborted',
'connection error',
'timeout',
'network error',
'temporary failure',
'rate limit'
]
is_retryable = any(err in error_msg for err in retryable_errors)
if not is_retryable or attempt == max_retries - 1:
raise e
wait_time = delay * (backoff ** attempt)
st.warning(f"⚠️ Attempt {attempt + 1} failed: {e}. Retrying in {wait_time:.1f}s...")
time.sleep(wait_time)
raise last_exception
return wrapper
return decorator
def handle_stream_error(func: Callable) -> Callable:
"""Handle stream buffer errors specifically"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
error_msg = str(e).lower()
if 'bodystreamBuffer' in error_msg or 'aborted' in error_msg:
st.error("🔄 Connection interrupted. Please try again.")
st.info("💡 Tip: Try refreshing the page if the issue persists.")
return None
else:
raise e
return wrapper
class ProgressTracker:
"""Track progress with better error handling"""
def __init__(self, total_jobs: int):
self.total_jobs = total_jobs
self.current_job = 0
self.progress_bar = None
self.status_text = None
def start(self):
"""Initialize progress tracking"""
self.progress_bar = st.progress(0)
self.status_text = st.empty()
def update(self, job_idx: int, status: str = "Processing..."):
"""Update progress"""
if self.progress_bar and self.status_text:
progress = (job_idx + 1) / self.total_jobs
self.progress_bar.progress(progress)
self.status_text.text(f"{status} ({job_idx + 1}/{self.total_jobs})")
def complete(self):
"""Mark as complete"""
if self.progress_bar and self.status_text:
self.progress_bar.progress(1.0)
self.status_text.text("✅ All jobs completed!") |