#!/usr/bin/env python3 """ 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}") # Quick connection test with timeout 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())