#!/usr/bin/env python3 """ SamCloud Ultimate Home Lab Dashboard A beautiful, Apple-like dashboard for monitoring your entire home lab infrastructure. Inspired by macOS design principles with real-time monitoring capabilities. """ 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) class UltimateHomeLab: def __init__(self): self.ddns_monitor_path = "/Volumes/SamMgmt/CF DDClient/ddns_monitor.py" self.start_time = datetime.now() # Service definitions with proper URLs and descriptions self.known_services = { 'dashy': { 'name': 'Dashy Dashboard', 'description': 'Home lab dashboard and service launcher', 'category': 'πŸ“Š Dashboards', 'icon': '🏠', 'color': '#667eea' }, 'homepage': { 'name': 'Homepage', 'description': 'Modern home lab dashboard', 'category': 'πŸ“Š Dashboards', 'icon': '🏑', 'color': '#4caf50' }, 'uptime-kuma': { 'name': 'Uptime Kuma', 'description': 'Uptime monitoring and status page', 'category': 'πŸ“Š Monitoring', 'icon': 'πŸ“ˆ', 'color': '#ff9800' }, 'openwebui': { 'name': 'Open WebUI', 'description': 'AI chat interface (Ollama frontend)', 'category': 'πŸ€– AI/ML', 'icon': '🧠', 'color': '#9c27b0' }, 'immich_server': { 'name': 'Immich Photos', 'description': 'Self-hosted photo and video management', 'category': 'πŸ“· Media', 'icon': 'πŸ“Έ', 'color': '#2196f3' }, 'npm': { 'name': 'Nginx Proxy Manager', 'description': 'Reverse proxy with SSL management', 'category': 'πŸ”§ Infrastructure', 'icon': '🌐', 'color': '#4caf50' }, 'jellyseerr': { 'name': 'Jellyseerr', 'description': 'Media request management for Plex/Jellyfin', 'category': '🎬 Media Automation', 'icon': '🎭', 'color': '#ff5722' }, 'radarr': { 'name': 'Radarr', 'description': 'Movie collection manager', 'category': '🎬 Media Automation', 'icon': '🎬', 'color': '#ffc107' }, 'sonarr': { 'name': 'Sonarr', 'description': 'TV series collection manager', 'category': '🎬 Media Automation', 'icon': 'πŸ“Ί', 'color': '#2196f3' }, 'lidarr': { 'name': 'Lidarr', 'description': 'Music collection manager', 'category': '🎬 Media Automation', 'icon': '🎡', 'color': '#9c27b0' }, 'bazarr': { 'name': 'Bazarr', 'description': 'Subtitle manager for Radarr and Sonarr', 'category': '🎬 Media Automation', 'icon': 'πŸ’¬', 'color': '#607d8b' }, 'prowlarr': { 'name': 'Prowlarr', 'description': 'Indexer manager for *arr suite', 'category': '🎬 Media Automation', 'icon': 'πŸ”', 'color': '#795548' }, 'jackett': { 'name': 'Jackett', 'description': 'Torrent indexer proxy', 'category': '🎬 Media Automation', 'icon': 'πŸ§₯', 'color': '#ff9800' }, 'qbittorrent': { 'name': 'qBittorrent', 'description': 'BitTorrent client', 'category': 'πŸ“₯ Downloads', 'icon': 'πŸ’Ύ', 'color': '#4caf50' }, 'navidrome': { 'name': 'Navidrome', 'description': 'Music streaming server', 'category': '🎡 Music', 'icon': '🎼', 'color': '#e91e63' }, 'filebrowser': { 'name': 'File Browser', 'description': 'Web-based file manager', 'category': 'πŸ—‚οΈ File Management', 'icon': 'πŸ“', 'color': '#607d8b' }, 'spotspot': { 'name': 'SpotSpot', 'description': 'Spotify playlist manager', 'category': '🎡 Music', 'icon': '🎧', 'color': '#1db954' } } def get_system_resources(self) -> Dict[str, Any]: """Get detailed system resource information""" try: # CPU information cpu_percent = psutil.cpu_percent(interval=1) cpu_count = psutil.cpu_count() cpu_freq = psutil.cpu_freq() load_avg = os.getloadavg() if hasattr(os, 'getloadavg') else [0, 0, 0] # Memory information memory = psutil.virtual_memory() # Disk information for main drive disk = psutil.disk_usage('/') # Network stats net_io = psutil.net_io_counters() # Boot time and uptime boot_time = datetime.fromtimestamp(psutil.boot_time()) uptime = datetime.now() - boot_time return { 'hostname': socket.gethostname(), 'platform': f"{platform.system()} {platform.release()}", 'architecture': platform.machine(), 'uptime': str(uptime).split('.')[0], # Remove microseconds 'boot_time': boot_time.isoformat(), 'cpu': { 'usage_percent': round(cpu_percent, 1), 'count': cpu_count, 'frequency_mhz': round(cpu_freq.current) if cpu_freq else 0, 'load_avg': [round(x, 2) for x in load_avg] }, 'memory': { 'total_gb': round(memory.total / (1024**3), 2), 'used_gb': round(memory.used / (1024**3), 2), 'available_gb': round(memory.available / (1024**3), 2), 'usage_percent': round(memory.percent, 1) }, 'disk': { 'total_gb': round(disk.total / (1024**3), 2), 'used_gb': round(disk.used / (1024**3), 2), 'free_gb': round(disk.free / (1024**3), 2), 'usage_percent': round((disk.used / disk.total) * 100, 1) }, 'network': { 'bytes_sent_mb': round(net_io.bytes_sent / (1024**2), 2), 'bytes_recv_mb': round(net_io.bytes_recv / (1024**2), 2), 'packets_sent': net_io.packets_sent, 'packets_recv': net_io.packets_recv } } except Exception as e: return {'error': str(e)} def get_docker_services(self) -> List[Dict[str, Any]]: """Get all Docker containers with enriched information""" services = [] try: result = subprocess.run( ['docker', 'ps', '-a', '--format', 'json'], capture_output=True, text=True, timeout=15 ) if result.returncode == 0: for line in result.stdout.strip().split('\n'): if line.strip(): container = json.loads(line) service = self._enrich_container_info(container) if service: # Filter out k8s system containers services.append(service) # Sort by category then name services.sort(key=lambda x: (x['category'], x['name'])) except Exception as e: print(f"Error getting Docker services: {e}") return services def _enrich_container_info(self, container: Dict) -> Optional[Dict[str, Any]]: """Enrich container information with service details""" name = container.get('Names', '') image = container.get('Image', '') status = container.get('Status', '') # Skip Kubernetes system containers if name.startswith('k8s_') and ('pause' in image or 'coredns' in name): return None # Get service info service_key = name.lower().replace('_', '-') service_info = self.known_services.get(service_key, {}) # Parse port mappings ports_raw = container.get('Ports', '') ports = self._parse_docker_ports(ports_raw) # Determine if service is healthy/running is_running = 'Up' in status is_healthy = 'healthy' in status.lower() or (is_running and 'unhealthy' not in status.lower()) # Get primary URL primary_url = None if ports: for port_info in ports: if port_info.get('external') and int(port_info['external']) < 10000: primary_url = f"http://localhost:{port_info['external']}" break return { 'name': service_info.get('name', name.title()), 'description': service_info.get('description', 'Container service'), 'category': service_info.get('category', '🐳 Other'), 'icon': service_info.get('icon', 'βš™οΈ'), 'color': service_info.get('color', '#607d8b'), 'container_name': name, 'image': image, 'status': status, 'ports': ports, 'is_running': is_running, 'is_healthy': is_healthy, 'url': primary_url, 'last_updated': datetime.now().isoformat() } def _parse_docker_ports(self, ports_str: str) -> List[Dict[str, str]]: """Parse Docker port mappings""" if not ports_str: return [] ports = [] for part in ports_str.split(', '): if '->' in part: external_part, internal_part = part.split('->') external_port = external_part.split(':')[-1] internal_port = internal_part.split('/')[0] ports.append({ 'external': external_port, 'internal': internal_port, 'protocol': 'tcp' }) return ports def get_ddns_status(self) -> Dict[str, Any]: """Get DDNS status from the existing monitor""" try: result = subprocess.run( ['python3', self.ddns_monitor_path, '--json'], capture_output=True, text=True, timeout=10 ) if result.returncode == 0: # Try to find JSON in the output for line in result.stdout.strip().split('\n'): line = line.strip() if line.startswith('{') and line.endswith('}'): try: return json.loads(line) except json.JSONDecodeError: continue return { 'status': 'healthy' if result.returncode == 0 else 'error', 'domain': 'wg.samcloud.ca', 'healthy': result.returncode == 0, 'last_check': datetime.now().isoformat() } except Exception as e: return { 'status': 'error', 'error': str(e), 'healthy': False, 'last_check': datetime.now().isoformat() } def get_network_overview(self) -> Dict[str, Any]: """Get network and port overview""" active_ports = [] try: # Get listening ports using netstat result = subprocess.run( ['netstat', '-an', '-p', 'tcp'], capture_output=True, text=True, timeout=10 ) if result.returncode == 0: ports_used = set() for line in result.stdout.split('\n'): if 'LISTEN' in line: parts = line.split() if len(parts) >= 4: addr_port = parts[3] try: if '.' in addr_port: port = int(addr_port.split('.')[-1]) if port not in ports_used and 20 <= port <= 65535: ports_used.add(port) active_ports.append({ 'port': port, 'type': self._categorize_port(port) }) except (ValueError, IndexError): continue active_ports.sort(key=lambda x: x['port']) except Exception as e: print(f"Network overview error: {e}") return { 'total_listening_ports': len(active_ports), 'web_services': len([p for p in active_ports if p['type'] == 'web']), 'system_services': len([p for p in active_ports if p['type'] == 'system']), 'user_services': len([p for p in active_ports if p['type'] == 'user']), 'active_ports': active_ports[:20] # Limit to first 20 for display } def _categorize_port(self, port: int) -> str: """Categorize port by type""" if port < 1024: return 'system' elif 8000 <= port <= 9999 or port in [80, 443, 3000, 3001, 4000, 5000]: return 'web' else: return 'user' def get_dashboard_data(self) -> Dict[str, Any]: """Get all dashboard data""" docker_services = self.get_docker_services() homebrew_services = self.get_homebrew_services() # Combine all services all_services = docker_services + homebrew_services return { 'system': self.get_system_resources(), 'services': all_services, 'homebrew_services': homebrew_services, 'docker_services': docker_services, 'ddns': self.get_ddns_status(), 'network': self.get_network_overview(), 'meta': { 'dashboard_uptime': str(datetime.now() - self.start_time).split('.')[0], 'last_refresh': datetime.now().isoformat(), 'version': '2.0.0' } } def get_homebrew_services(self) -> List[Dict[str, Any]]: """Get Homebrew services with enhanced error detection""" services = [] try: result = subprocess.run( ['brew', 'services', 'list'], capture_output=True, text=True, timeout=10 ) if result.returncode == 0: lines = result.stdout.strip().split('\n')[1:] # Skip header for line in lines: parts = line.split() if len(parts) >= 3: name = parts[0] status = parts[1] # Enhanced status detection and log capture service_logs = [] # Check if process is actually running (for cases like dnsmasq) is_process_running = self._check_process_running(name) # Determine health status with better logic if status == 'started' or (status == 'error' and is_process_running): service_status = 'running' health_score = 100 is_healthy = True is_running = True if status == 'error' and is_process_running: service_logs.append(f"Service reports error status but process is running") elif status == 'stopped': service_status = 'stopped' health_score = 0 is_healthy = False is_running = False elif 'error' in status: service_status = 'error' health_score = 10 is_healthy = False is_running = False # Get detailed error logs error_logs = self._get_service_error_logs(name) service_logs.extend(error_logs) else: service_status = status health_score = 50 is_healthy = False is_running = False # Get service details service_info = self._get_homebrew_service_info(name) service = { 'name': service_info.get('display_name', name.title()), 'container_name': name, 'description': service_info.get('description', f'Homebrew service: {name}'), 'category': service_info.get('category', '🍺 Homebrew Services'), 'icon': service_info.get('icon', '🍺'), 'color': service_info.get('color', '#fbb040'), 'status': f'{service_status.title()} ({status})', 'is_running': is_running, 'is_healthy': is_healthy, 'url': service_info.get('url'), 'ports': service_info.get('ports', []), 'image': 'homebrew', 'logs': service_logs, 'last_updated': datetime.now().isoformat() } services.append(service) except Exception as e: print(f"Error getting Homebrew services: {e}") return services def _check_process_running(self, service_name: str) -> bool: """Check if a process is actually running by name""" try: result = subprocess.run( ['pgrep', '-f', service_name], capture_output=True, text=True, timeout=5 ) return result.returncode == 0 and bool(result.stdout.strip()) except: return False def _get_service_error_logs(self, service_name: str) -> List[str]: """Get error logs for a specific service""" logs = [] # Common log locations log_paths = [ f'/opt/homebrew/var/log/{service_name}.log', f'/opt/homebrew/var/log/{service_name}/error.log', f'/usr/local/var/log/{service_name}.log', f'/usr/local/var/log/{service_name}/error.log' ] # Service-specific log locations if service_name == 'nginx': log_paths.extend([ '/opt/homebrew/var/log/nginx/error.log', '/usr/local/var/log/nginx/error.log' ]) elif service_name == 'dnsmasq': log_paths.extend([ '/opt/homebrew/var/log/dnsmasq.log', '/usr/local/var/log/dnsmasq.log' ]) for log_path in log_paths: try: if os.path.exists(log_path): result = subprocess.run( ['tail', '-3', log_path], capture_output=True, text=True, timeout=5 ) if result.returncode == 0 and result.stdout.strip(): log_lines = result.stdout.strip().split('\n') for line in log_lines: if 'error' in line.lower() or 'failed' in line.lower() or 'bind()' in line.lower(): logs.append(line.strip()) break except: continue # Special handling for nginx port conflicts if service_name == 'nginx' and not logs: try: result = subprocess.run( ['lsof', '-i', ':80'], capture_output=True, text=True, timeout=5 ) if result.returncode == 0: port_users = result.stdout.strip().split('\n')[1:] # Skip header if port_users: logs.append("Port 80 conflict detected:") for user in port_users[:2]: # Show first 2 logs.append(f" {user.split()[0]} (PID {user.split()[1]})") except: logs.append("Unable to check port 80 status") return logs[:5] # Limit to 5 recent log entries def _get_homebrew_service_info(self, service_name: str) -> Dict[str, Any]: """Get enhanced service information for homebrew services""" service_definitions = { 'nginx': { 'display_name': 'Nginx', 'category': 'πŸ”§ Infrastructure', 'icon': '🌐', 'color': '#009639', 'description': 'High-performance web server and reverse proxy', 'url': 'http://localhost:80', 'ports': [{'external': '80', 'internal': '80', 'protocol': 'HTTP'}, {'external': '443', 'internal': '443', 'protocol': 'HTTPS'}] }, 'dnsmasq': { 'display_name': 'DNSmasq', 'category': '🌐 Network Services', 'icon': 'πŸ—‚οΈ', 'color': '#2196f3', 'description': 'Lightweight DNS forwarder and DHCP server', 'url': None, 'ports': [{'external': '53', 'internal': '53', 'protocol': 'DNS'}] }, 'caddy': { 'display_name': 'Caddy', 'category': 'πŸ”§ Infrastructure', 'icon': 'πŸ•ΈοΈ', 'color': '#1f88e5', 'description': 'Modern web server with automatic HTTPS', 'url': 'http://home.samcloud.ca', 'ports': [{'external': '80', 'internal': '80', 'protocol': 'HTTP'}, {'external': '443', 'internal': '443', 'protocol': 'HTTPS'}] }, 'ddclient': { 'display_name': 'DDClient', 'category': '🌐 Network Services', 'icon': '🌍', 'color': '#4caf50', 'description': 'Dynamic DNS client for domain updates', 'url': None, 'ports': [] }, 'ollama': { 'display_name': 'Ollama', 'category': 'πŸ€– AI/ML', 'icon': 'πŸ¦™', 'color': '#ff6b35', 'description': 'Local LLM server for running AI models', 'url': 'http://localhost:11434', 'ports': [{'external': '11434', 'internal': '11434', 'protocol': 'HTTP'}] }, 'postgresql': { 'display_name': 'PostgreSQL', 'category': 'πŸ”§ Infrastructure', 'icon': '🐘', 'color': '#336791', 'description': 'Advanced open-source relational database', 'url': None, 'ports': [{'external': '5432', 'internal': '5432', 'protocol': 'PostgreSQL'}] } } return service_definitions.get(service_name, { 'display_name': service_name.title(), 'category': '🍺 Homebrew Services', 'icon': '🍺', 'color': '#fbb040', 'description': f'Homebrew service: {service_name}', 'url': None, 'ports': [] }) def get_all_services(self) -> List[Dict[str, Any]]: """Get all services including Docker and Homebrew""" all_services = self.get_docker_services() homebrew_services = self.get_homebrew_services() # Combine and remove duplicates service_names = set() for service in all_services + homebrew_services: if service['name'] not in service_names: service_names.add(service['name']) # Enrich Homebrew services with Docker-like structure enriched_homebrew_services = [] for name in service_names: docker_service = next((s for s in all_services if s['name'] == name), None) homebrew_service = next((s for s in homebrew_services if s['name'] == name), None) if docker_service: enriched_homebrew_services.append(docker_service) elif homebrew_service: enriched_homebrew_services.append({ 'name': homebrew_service['name'], 'description': homebrew_service['description'], 'category': homebrew_service['category'], 'icon': homebrew_service['icon'], 'color': homebrew_service['color'], 'container_name': homebrew_service['container_name'], 'image': homebrew_service['image'], 'status': homebrew_service['status'], 'ports': homebrew_service['ports'], 'is_running': homebrew_service['is_running'], 'is_healthy': homebrew_service['is_healthy'], 'url': homebrew_service['url'], 'last_updated': homebrew_service['last_updated'], 'logs': homebrew_service['logs'] }) return enriched_homebrew_services # Create dashboard instance homelab = UltimateHomeLab() # Modern Apple-like dashboard template DASHBOARD_TEMPLATE = """ SamCloud Home Lab

🏠 SamCloud

Home Lab Dashboard

Loading dashboard data...
""" @app.route('/') def dashboard(): """Main dashboard page""" return DASHBOARD_TEMPLATE @app.route('/api/ultimate') def api_ultimate(): """Ultimate dashboard API endpoint""" return jsonify(homelab.get_dashboard_data()) @app.route('/api/system') def api_system(): """System resources API""" return jsonify(homelab.get_system_resources()) @app.route('/api/services') def api_services(): """Docker services API""" return jsonify(homelab.get_docker_services()) @app.route('/api/ddns') def api_ddns(): """DDNS status API""" return jsonify(homelab.get_ddns_status()) @app.route('/api/network') def api_network(): """Network overview API""" return jsonify(homelab.get_network_overview()) @app.route('/health') def health(): """Health check""" return {'status': 'healthy', 'timestamp': datetime.now().isoformat()} if __name__ == '__main__': import argparse parser = argparse.ArgumentParser(description='SamCloud Ultimate 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 Ultimate Dashboard on http://{args.host}:{args.port}") print("✨ Features:") print(" β€’ Modern Apple-like design") print(" β€’ Real-time system monitoring") print(" β€’ Smart service detection") print(" β€’ Clickable service links") print(" β€’ Auto-refresh capabilities") print(" β€’ Mobile responsive design") app.run(host=args.host, port=args.port, debug=args.debug)