| |
| """ |
| Quick test for WebSocket retry timing - reduced attempts for faster results. |
| """ |
|
|
| import asyncio |
| import websockets |
| import time |
| from datetime import datetime |
|
|
| async def quick_retry_test(): |
| """Quick test with just 3 attempts to see timing patterns.""" |
| print(f"🎯 QUICK RETRY TEST - {datetime.now().strftime('%H:%M:%S')}") |
| |
| ws_url = "wss://pgits-stt-gpu-service.hf.space/ws/stt" |
| max_retries = 3 |
| base_delay = 2.0 |
| backoff_multiplier = 1.5 |
| |
| start_time = time.time() |
| |
| for attempt in range(1, max_retries + 1): |
| attempt_start = time.time() |
| |
| if attempt > 1: |
| delay = base_delay * (backoff_multiplier ** (attempt - 2)) |
| print(f"⏰ Waiting {delay:.1f}s before attempt {attempt}") |
| await asyncio.sleep(delay) |
| |
| try: |
| print(f"🎤 Attempt {attempt}/{max_retries} - Connecting to {ws_url}") |
| |
| |
| async with asyncio.timeout(10): |
| async with websockets.connect(ws_url) as websocket: |
| connection_time = time.time() - attempt_start |
| print(f"✅ Connected in {connection_time:.1f}s") |
| return |
| |
| except websockets.exceptions.InvalidStatusCode as e: |
| connection_time = time.time() - attempt_start |
| status_code = getattr(e, 'status_code', 'unknown') |
| print(f"🚫 HTTP {status_code} after {connection_time:.1f}s") |
| |
| if status_code == 503: |
| print(" → Service cold starting") |
| elif status_code == 403: |
| print(" → WebSocket endpoint not available") |
| |
| except Exception as e: |
| connection_time = time.time() - attempt_start |
| print(f"❌ {type(e).__name__}: {e} after {connection_time:.1f}s") |
| |
| total_time = time.time() - start_time |
| print(f"🏁 Test completed in {total_time:.1f}s total") |
|
|
| if __name__ == "__main__": |
| asyncio.run(quick_retry_test()) |
