import requestsimport timedef ejecutar_busqueda_roku(ip, app_name, query): """Execute search in Roku app.""" # Get app ID from mapping app_id = ROKU_APPS.get(app_name.lower(), "837") # Default to YouTube if app_id: # Launch the app enviar_comando_roku(ip, f"launch/{app_id}") time.sleep(2) # Wait for app to load # URL encode the query query_encoded = requests.utils.quote(query) # Execute search endpoint = f"search/browse?keyword={query_encoded}&launch=true&provider-id={app_id}" return enviar_comando_roku(ip, endpoint) return False# Examplesip = "192.168.1.8"# Search YouTubeejecutar_busqueda_roku(ip, "youtube", "recetas de cocina")# Search Netflixejecutar_busqueda_roku(ip, "netflix", "películas de acción")# Search with special charactersejecutar_busqueda_roku(ip, "prime", "El Señor de los Anillos")
List of discovered devices with ‘ip’ and ‘tipo’ keys
import socketimport threadingimport timedef escanear_red_tvs(target_ip="192.168.1.8"): """Scan network for Roku devices.""" encontrados = [] # Try target IP first tv_principal = verificar_roku_por_info(target_ip) if tv_principal: encontrados.append(tv_principal) return encontrados try: # Get local IP to determine network range s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) ip_local = s.getsockname()[0] s.close() # Extract network prefix (e.g., "192.168.1") prefijo = ".".join(ip_local.split(".")[:-1]) # Scan function for threading def tarea_escaneo(ip_test): res = verificar_roku_por_info(ip_test) if res: encontrados.append(res) # Scan all IPs in range hilos = [] for i in range(1, 255): t = threading.Thread( target=tarea_escaneo, args=(f"{prefijo}.{i}",) ) t.daemon = True t.start() hilos.append(t) # Wait for scan to complete time.sleep(1.5) except: pass return encontrados# Usagedevices = escanear_red_tvs()print(f"Found {len(devices)} Roku device(s):")for device in devices: print(f" {device['tipo']} at {device['ip']}")
import requestsfrom requests.exceptions import Timeout, ConnectionErrordef safe_roku_command(ip, endpoint, retries=3): """Execute Roku command with retry logic.""" for attempt in range(retries): try: url = f"http://{ip}:8060/{endpoint}" response = requests.post(url, timeout=5) return True except Timeout: print(f"Timeout on attempt {attempt + 1}") except ConnectionError: print(f"Connection failed on attempt {attempt + 1}") except Exception as e: print(f"Unexpected error: {e}") return False
Audio Errors
import pygamefrom gtts import gTTSdef safe_speak(text): """TTS with error handling.""" try: tts = gTTS(text=text, lang='es') tts.save('temp.mp3') pygame.mixer.init() pygame.mixer.music.load('temp.mp3') pygame.mixer.music.play() while pygame.mixer.music.get_busy(): time.sleep(0.1) except ImportError: print("gTTS or pygame not installed") except Exception as e: print(f"TTS error: {e}") finally: try: pygame.mixer.music.unload() if os.path.exists('temp.mp3'): os.remove('temp.mp3') except: pass
Device Discovery
def safe_network_scan(timeout=2.0): """Network scan with error handling.""" devices = [] try: # Try to get local IP s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.settimeout(timeout) s.connect(("8.8.8.8", 80)) local_ip = s.getsockname()[0] s.close() prefix = ".".join(local_ip.split(".")[:-1]) # Scan network for i in range(1, 255): ip = f"{prefix}.{i}" device = verificar_roku_por_info(ip) if device: devices.append(device) except socket.timeout: print("Network timeout during scan") except socket.error as e: print(f"Socket error: {e}") except Exception as e: print(f"Scan error: {e}") return devices
