#!/usr/bin/env python3
"""
SamCloud Home Server Dashboard
A comprehensive dashboard for monitoring your home lab infrastructure:
- DDNS status and management
- System resources (CPU, memory, disk)
- Network monitoring
- Service status (containers, processes)
- Beautiful, responsive UI inspired by Dashy/Homepage
- API endpoints for integration with other dashboards
Usage:
python3 homelab_dashboard.py --host 0.0.0.0 --port 8080
"""
import json
import os
import platform
import psutil
import requests
import socket
import subprocess
import time
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, List, Optional, Any
from flask import Flask, render_template_string, jsonify, request
from flask_cors import CORS
app = Flask(__name__)
CORS(app) # Enable CORS for API access
class HomeLabDashboard:
def __init__(self):
self.ddns_monitor_path = "/Volumes/SamMgmt/CF DDClient/ddns_monitor.py"
self.start_time = datetime.now()
def get_ddns_status(self) -> Dict[str, Any]:
"""Get DDNS status from our existing monitor"""
try:
result = subprocess.run(
['python3', self.ddns_monitor_path, '--json'],
capture_output=True, text=True, timeout=30
)
if result.returncode == 0:
# Parse the JSON output - look for the last valid JSON object
lines = result.stdout.strip().split('\n')
json_data = None
for line in reversed(lines): # Start from the end
line = line.strip()
if line.startswith('{') and line.endswith('}'):
try:
json_data = json.loads(line)
break
except json.JSONDecodeError:
continue
if json_data:
return json_data
else:
# Try to extract from the output manually
output = result.stdout.strip()
if 'healthy":true' in output or 'healthy": true' in output:
return {
"status": "ok",
"healthy": True,
"domain": "wg.samcloud.ca",
"note": "Parsed from monitor output"
}
return {
"status": "error",
"error": "Failed to parse DDNS monitor output",
"healthy": False,
"raw_output": result.stdout[:200] if result.stdout else "No output"
}
except Exception as e:
return {
"status": "error",
"error": str(e),
"healthy": False
}
def get_system_info(self) -> Dict[str, Any]:
"""Get system information and resources"""
try:
# Basic system info
system_info = {
"hostname": socket.gethostname(),
"platform": platform.system(),
"platform_release": platform.release(),
"architecture": platform.machine(),
"uptime": str(datetime.now() - datetime.fromtimestamp(psutil.boot_time())),
"dashboard_uptime": str(datetime.now() - self.start_time)
}
# CPU information
cpu_info = {
"usage": psutil.cpu_percent(interval=1),
"count": psutil.cpu_count(),
"load_avg": os.getloadavg() if hasattr(os, 'getloadavg') else [0, 0, 0]
}
# Memory information
memory = psutil.virtual_memory()
memory_info = {
"total": memory.total,
"available": memory.available,
"used": memory.used,
"percentage": memory.percent,
"total_gb": round(memory.total / (1024**3), 2),
"used_gb": round(memory.used / (1024**3), 2),
"available_gb": round(memory.available / (1024**3), 2)
}
# Disk information
disk_usage = psutil.disk_usage('/')
disk_info = {
"total": disk_usage.total,
"used": disk_usage.used,
"free": disk_usage.free,
"percentage": (disk_usage.used / disk_usage.total) * 100,
"total_gb": round(disk_usage.total / (1024**3), 2),
"used_gb": round(disk_usage.used / (1024**3), 2),
"free_gb": round(disk_usage.free / (1024**3), 2)
}
# Network information
network_stats = psutil.net_io_counters()
network_info = {
"bytes_sent": network_stats.bytes_sent,
"bytes_recv": network_stats.bytes_recv,
"packets_sent": network_stats.packets_sent,
"packets_recv": network_stats.packets_recv,
"bytes_sent_mb": round(network_stats.bytes_sent / (1024**2), 2),
"bytes_recv_mb": round(network_stats.bytes_recv / (1024**2), 2)
}
return {
"system": system_info,
"cpu": cpu_info,
"memory": memory_info,
"disk": disk_info,
"network": network_info,
"timestamp": datetime.now().isoformat()
}
except Exception as e:
return {"error": str(e)}
def get_running_services(self) -> List[Dict[str, Any]]:
"""Get running services and processes"""
services = []
# Check for common home lab services
common_services = {
"ddclient": "DDNS Client",
"nginx": "Web Server",
"docker": "Docker",
"ssh": "SSH Server",
"smbd": "Samba",
"plex": "Plex Media Server",
"pihole": "Pi-hole",
"homebridge": "Homebridge",
"node_exporter": "Prometheus Node Exporter",
"grafana": "Grafana",
"prometheus": "Prometheus"
}
for service_name, display_name in common_services.items():
try:
# Check if process is running
for process in psutil.process_iter(['pid', 'name', 'cmdline']):
if any(service_name in str(item).lower() for item in process.info['cmdline'] or []):
services.append({
"name": display_name,
"status": "running",
"pid": process.info['pid'],
"healthy": True
})
break
else:
# Process not found
services.append({
"name": display_name,
"status": "stopped",
"pid": None,
"healthy": False
})
except Exception:
services.append({
"name": display_name,
"status": "unknown",
"pid": None,
"healthy": None
})
return services
def get_docker_containers(self) -> List[Dict[str, Any]]:
"""Get Docker container status with detailed information"""
containers = []
try:
# Get detailed container info
result = subprocess.run(
['docker', 'ps', '--format', 'json'],
capture_output=True, text=True, timeout=10
)
if result.returncode == 0:
for line in result.stdout.strip().split('\n'):
if line.strip():
container = json.loads(line)
# Extract port mappings
ports_raw = container.get("Ports", "")
port_mappings = self._parse_port_mappings(ports_raw)
# Get container name (remove k8s prefix if present)
name = container.get("Names", "unknown")
display_name = self._get_service_display_name(name)
# Determine service category/type
service_type = self._get_service_type(name, container.get("Image", ""))
containers.append({
"name": display_name,
"container_name": name,
"image": container.get("Image", "unknown"),
"status": container.get("Status", "unknown"),
"ports": port_mappings,
"service_type": service_type,
"healthy": "Up" in container.get("Status", "") and "healthy" in container.get("Status", "").lower(),
"url": self._get_service_url(name, port_mappings)
})
# Sort by service type then name
containers.sort(key=lambda x: (x['service_type'], x['name']))
except Exception as e:
print(f"Error getting containers: {e}")
return containers
def _parse_port_mappings(self, ports_str: str) -> List[Dict[str, str]]:
"""Parse Docker port mappings from ports string"""
mappings = []
if not ports_str:
return mappings
# Handle multiple port mappings separated by comma
for port_part in ports_str.split(', '):
if '->' in port_part:
# External -> Internal mapping
external, internal = port_part.split('->')
external_port = external.split(':')[-1] if ':' in external else external
internal_port = internal.split('/')[0] if '/' in internal else internal
mappings.append({
"external": external_port,
"internal": internal_port,
"protocol": "tcp" # Default
})
elif '/' in port_part:
# Just internal port with protocol
port, protocol = port_part.split('/')
mappings.append({
"external": None,
"internal": port,
"protocol": protocol
})
return mappings
def _get_service_display_name(self, container_name: str) -> str:
"""Get a friendly display name for the service"""
# Remove k8s prefixes
if container_name.startswith('k8s_'):
parts = container_name.split('_')
if len(parts) > 1:
return parts[1]
# Known service mappings
service_names = {
'uptime-kuma': 'Uptime Kuma',
'openwebui': 'Open WebUI',
'immich_server': 'Immich Photos',
'immich_machine_learning': 'Immich ML',
'immich_redis': 'Immich Redis',
'immich_postgres': 'Immich Database',
'filebrowser': 'File Browser',
'npm': 'Nginx Proxy Manager',
'sportyfin': 'SportyFin',
'jellyseerr': 'Jellyseerr',
'qbittorrent': 'qBittorrent',
'spotspot': 'SpotSpot',
'spotizerr-redis': 'Spotizerr Redis',
'navidrome': 'Navidrome Music',
'dashy': 'Dashy Dashboard',
'homepage': 'Homepage Dashboard'
}
return service_names.get(container_name, container_name.title())
def _get_service_type(self, name: str, image: str) -> str:
"""Categorize the service type"""
media_services = ['radarr', 'sonarr', 'lidarr', 'bazarr', 'prowlarr', 'jackett', 'qbittorrent', 'jellyseerr']
dashboard_services = ['dashy', 'homepage', 'uptime-kuma']
media_servers = ['navidrome', 'immich', 'spotspot']
infrastructure = ['npm', 'redis', 'postgres', 'filebrowser']
ai_services = ['openwebui']
name_lower = name.lower()
if any(service in name_lower for service in media_services):
return "πΊ Media"
elif any(service in name_lower for service in dashboard_services):
return "π Dashboard"
elif any(service in name_lower for service in media_servers):
return "π΅ Media Server"
elif any(service in name_lower for service in infrastructure):
return "π§ Infrastructure"
elif any(service in name_lower for service in ai_services):
return "π€ AI/ML"
elif 'k8s_' in name:
return "βΈοΈ Kubernetes"
else:
return "π³ Other"
def _get_service_url(self, name: str, port_mappings: List[Dict]) -> Optional[str]:
"""Get the local URL for the service if accessible"""
for mapping in port_mappings:
if mapping.get('external'):
port = mapping['external']
# Skip if it's just a number without clear protocol
try:
port_num = int(port)
if 80 <= port_num <= 9999: # Likely web services
return f"http://localhost:{port}"
except ValueError:
pass
return None
def get_network_ports(self) -> List[Dict[str, Any]]:
"""Get all listening ports on the system using netstat"""
ports = []
try:
# Use netstat command which doesn't require special permissions
result = subprocess.run(
['netstat', '-an', '-p', 'tcp'],
capture_output=True, text=True, timeout=10
)
if result.returncode == 0:
lines = result.stdout.strip().split('\n')
port_set = set() # To avoid duplicates
for line in lines:
if 'LISTEN' in line:
parts = line.split()
if len(parts) >= 4:
address_port = parts[3]
if '.' in address_port:
try:
# Parse address:port
if address_port.count('.') >= 3: # IPv4
# Handle IPv4 addresses like 127.0.0.1.7429
parts_addr = address_port.split('.')
port = int(parts_addr[-1])
address = '.'.join(parts_addr[:-1])
else:
# Handle cases like *.80
if ':' in address_port:
address, port = address_port.rsplit(':', 1)
else:
address, port = address_port.rsplit('.', 1)
port = int(port)
if port not in port_set:
port_set.add(port)
# Try to get process info using lsof
process_info = self._get_process_for_port(port)
ports.append({
"port": port,
"address": address if address != '*' else '0.0.0.0',
"process_name": process_info.get('name', 'Unknown'),
"process_cmdline": process_info.get('cmdline', 'Unknown'),
"pid": process_info.get('pid')
})
except (ValueError, IndexError):
continue
# Sort by port number
ports.sort(key=lambda x: x['port'])
except Exception as e:
print(f"Error getting network ports: {e}")
# Fallback: add at least our own dashboard port
ports = [{
"port": 7429,
"address": "0.0.0.0",
"process_name": "homelab_dashboard.py",
"process_cmdline": "python3 homelab_dashboard.py",
"pid": os.getpid()
}]
return ports
def _get_process_for_port(self, port: int) -> Dict[str, Any]:
"""Get process information for a specific port using lsof"""
try:
result = subprocess.run(
['lsof', '-i', f':{port}', '-P', '-n'],
capture_output=True, text=True, timeout=5
)
if result.returncode == 0:
lines = result.stdout.strip().split('\n')
for line in lines[1:]: # Skip header
parts = line.split(None, 8)
if len(parts) >= 8:
return {
'name': parts[0],
'pid': int(parts[1]) if parts[1].isdigit() else None,
'cmdline': parts[0] # lsof doesn't give full cmdline
}
except Exception:
pass
return {'name': 'Unknown', 'pid': None, 'cmdline': 'Unknown'}
def get_dashboard_data(self) -> Dict[str, Any]:
"""Get all dashboard data"""
return {
"ddns": self.get_ddns_status(),
"system": self.get_system_info(),
"services": self.get_running_services(),
"containers": self.get_docker_containers(),
"ports": self.get_network_ports(),
"timestamp": datetime.now().isoformat()
}
# Dashboard HTML template
DASHBOARD_HTML = """
SamCloud Home Lab Dashboard
"""
# Create dashboard instance
dashboard = HomeLabDashboard()
@app.route('/')
def index():
"""Main dashboard page"""
return DASHBOARD_HTML
@app.route('/api/dashboard')
def api_dashboard():
"""API endpoint for dashboard data"""
return jsonify(dashboard.get_dashboard_data())
@app.route('/api/ddns')
def api_ddns():
"""API endpoint for DDNS status only"""
return jsonify(dashboard.get_ddns_status())
@app.route('/api/system')
def api_system():
"""API endpoint for system info only"""
return jsonify(dashboard.get_system_info())
@app.route('/api/services')
def api_services():
"""API endpoint for services only"""
return jsonify(dashboard.get_running_services())
@app.route('/api/containers')
def api_containers():
"""API endpoint for Docker containers only"""
return jsonify(dashboard.get_docker_containers())
@app.route('/api/ports')
def api_ports():
"""API endpoint for network ports only"""
return jsonify(dashboard.get_network_ports())
@app.route('/health')
def health():
"""Health check endpoint"""
return {'status': 'ok', 'timestamp': datetime.now().isoformat()}
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(description='SamCloud Home Lab Dashboard')
parser.add_argument('--host', default='0.0.0.0', help='Host to bind to')
parser.add_argument('--port', type=int, default=7429, help='Port to bind to')
parser.add_argument('--debug', action='store_true', help='Enable debug mode')
args = parser.parse_args()
print(f"π Starting SamCloud Home Lab Dashboard on http://{args.host}:{args.port}")
print("π Dashboard features:")
print(" β’ DDNS monitoring (using your existing setup)")
print(" β’ System resource monitoring")
print(" β’ Service status monitoring")
print(" β’ Docker container monitoring")
print(" β’ REST API endpoints")
print(" β’ Auto-refresh every 30 seconds")
app.run(host=args.host, port=args.port, debug=args.debug)