#!/usr/bin/env python3 """ Hyprmonitors Home Assistant Connectivity Test This script helps diagnose connectivity issues between Home Assistant and the Hyprmonitors server, particularly for Docker setups. Usage: python test_connectivity.py [host] [port] Examples: python test_connectivity.py python test_connectivity.py localhost 3000 python test_connectivity.py host.docker.internal 3000 python test_connectivity.py 192.168.1.100 3000 """ import asyncio import aiohttp import sys import time from typing import Optional, Tuple async def test_connection(host: str = "localhost", port: int = 3000) -> bool: """Test connection to Hyprmonitors server.""" url = f"http://{host}:{port}/health" print(f"Testing connection to {url}...") try: timeout = aiohttp.ClientTimeout(total=10) async with aiohttp.ClientSession(timeout=timeout) as session: start_time = time.time() async with session.get(url) as response: end_time = time.time() response_time = (end_time - start_time) * 1000 # Convert to ms print(f"✓ Connection successful! Response time: {response_time:.2f}ms") print(f" Status: {response.status}") print(f" Content-Type: {response.headers.get('content-type', 'unknown')}") if response.status == 200: try: data = await response.json() print(f" Response: {data}") if data.get("success"): print("✓ Health check passed!") return True else: print("✗ Health check failed - server returned success=false") return False except Exception as e: print(f"✗ Failed to parse JSON response: {e}") text = await response.text() print(f" Raw response: {text[:200]}...") return False else: print(f"✗ Server returned status {response.status}") return False except asyncio.TimeoutError: print("✗ Connection timed out (10 seconds)") print(" This usually means:") print(" - The server is not running") print(" - Wrong host/port configuration") print(" - Network/firewall blocking the connection") return False except aiohttp.ClientConnectorError as e: print(f"✗ Connection failed: {e}") print(" This usually means:") print(" - The server is not running") print(" - Wrong host/port configuration") print(" - If using Docker: try 'host.docker.internal' instead of 'localhost'") return False except Exception as e: print(f"✗ Unexpected error: {e}") return False async def test_monitor_status(host: str = "localhost", port: int = 3000) -> bool: """Test monitor status endpoint.""" url = f"http://{host}:{port}/monitors/status" print(f"\nTesting monitor status endpoint: {url}...") try: timeout = aiohttp.ClientTimeout(total=10) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.get(url) as response: if response.status == 200: try: data = await response.json() print(f"✓ Monitor status retrieved successfully!") print(f" Success: {data.get('success')}") monitors = data.get('monitors', {}) print(f" Found {len(monitors)} monitors:") for monitor, status in monitors.items(): print(f" - {monitor}: {status}") return True except Exception as e: print(f"✗ Failed to parse monitor status response: {e}") return False else: print(f"✗ Monitor status request failed with status {response.status}") return False except Exception as e: print(f"✗ Monitor status test failed: {e}") return False def test_docker_alternatives(port: int = 3000) -> list: """Test common Docker host alternatives.""" alternatives = [ "host.docker.internal", "172.17.0.1", "docker.host.internal", "gateway.docker.internal" ] print("\n" + "="*60) print("TESTING DOCKER HOST ALTERNATIVES") print("="*60) working_hosts = [] for host in alternatives: print(f"\nTesting {host}:{port}...") try: result = asyncio.run(test_connection(host, port)) if result: working_hosts.append(host) print(f"✓ {host} works!") else: print(f"✗ {host} failed") except Exception as e: print(f"✗ {host} error: {e}") return working_hosts def print_summary(host: str, port: int, health_ok: bool, status_ok: bool): """Print test summary and recommendations.""" print("\n" + "="*60) print("TEST SUMMARY") print("="*60) print(f"Host: {host}") print(f"Port: {port}") print(f"Health check: {'✓ PASS' if health_ok else '✗ FAIL'}") print(f"Monitor status: {'✓ PASS' if status_ok else '✗ FAIL'}") if health_ok and status_ok: print("\n🎉 All tests passed! You can use these settings in Home Assistant:") print(f" Host: {host}") print(f" Port: {port}") else: print("\n❌ Tests failed. Try these troubleshooting steps:") print("\n1. Make sure the Hyprmonitors server is running:") print(" systemctl --user status hyprmonitors") print(" # or if using cargo:") print(" cargo run --release") print("\n2. Test locally on the host machine:") print(f" curl http://localhost:{port}/health") print("\n3. If Home Assistant is in Docker, try:") print(" - Use 'host.docker.internal' instead of 'localhost'") print(" - Use your host machine's IP address") print(" - Add '--network host' to Docker run command") print("\n4. Check firewall settings:") print(f" sudo ufw allow {port}") print(f" # or check iptables rules") if host == "localhost": print("\n5. For Docker users, run this script with Docker alternatives:") print(" python test_connectivity.py host.docker.internal 3000") async def main(): """Main function.""" # Parse command line arguments host = "localhost" port = 3000 if len(sys.argv) > 1: host = sys.argv[1] if len(sys.argv) > 2: try: port = int(sys.argv[2]) except ValueError: print(f"Invalid port: {sys.argv[2]}") sys.exit(1) print("Hyprmonitors Home Assistant Connectivity Test") print("="*50) # Test basic connectivity and health health_ok = await test_connection(host, port) # Test monitor status endpoint status_ok = False if health_ok: status_ok = await test_monitor_status(host, port) # If localhost failed and we might be in Docker, test alternatives if not health_ok and host == "localhost": print(f"\nLocalhost failed. Testing Docker alternatives...") working_hosts = test_docker_alternatives(port) if working_hosts: print(f"\n✓ Working Docker hosts found: {working_hosts}") print(f"Recommendation: Use '{working_hosts[0]}' as your host in Home Assistant") # Test the first working host print(f"\nTesting monitor status with {working_hosts[0]}...") status_ok = await test_monitor_status(working_hosts[0], port) host = working_hosts[0] # Update for summary health_ok = True # We know it works # Print summary and recommendations print_summary(host, port, health_ok, status_ok) # Exit with appropriate code sys.exit(0 if (health_ok and status_ok) else 1) if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: print("\n\nTest cancelled by user") sys.exit(130) except Exception as e: print(f"\nUnexpected error: {e}") sys.exit(1)