feat: Added stuff
This commit is contained in:
uttarayan21
239
homeassistant/test_connectivity.py
Normal file
239
homeassistant/test_connectivity.py
Normal file
@@ -0,0 +1,239 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user