409 lines
12 KiB
Python
409 lines
12 KiB
Python
"""
|
|
System API for health checks, system information, and diagnostics.
|
|
"""
|
|
|
|
import logging
|
|
import platform
|
|
import psutil
|
|
import os
|
|
from typing import Dict, Any, Optional
|
|
from datetime import datetime, timedelta
|
|
from fastapi import APIRouter, HTTPException
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from ..core.config import get_config
|
|
from ..core.input_ctrl import is_gspro_running, get_gspro_process_info
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter()
|
|
|
|
# Track application start time
|
|
APP_START_TIME = datetime.now()
|
|
|
|
|
|
@router.get("/health")
|
|
async def health_check():
|
|
"""
|
|
Health check endpoint for monitoring.
|
|
|
|
Returns basic health status and service availability.
|
|
"""
|
|
config = get_config()
|
|
|
|
# Check GSPro status
|
|
gspro_running = is_gspro_running(config.gspro.window_title)
|
|
|
|
# Calculate uptime
|
|
uptime = datetime.now() - APP_START_TIME
|
|
|
|
return {
|
|
"status": "healthy",
|
|
"timestamp": datetime.now().isoformat(),
|
|
"uptime_seconds": int(uptime.total_seconds()),
|
|
"services": {
|
|
"api": "running",
|
|
"gspro": "connected" if gspro_running else "disconnected",
|
|
"mdns": "enabled" if config.server.mdns_enabled else "disabled",
|
|
},
|
|
}
|
|
|
|
|
|
@router.get("/info")
|
|
async def system_info():
|
|
"""
|
|
Get detailed system information.
|
|
|
|
Returns information about the host system, Python environment, and application.
|
|
"""
|
|
config = get_config()
|
|
|
|
# Get system information
|
|
system_info = {
|
|
"platform": {
|
|
"system": platform.system(),
|
|
"release": platform.release(),
|
|
"version": platform.version(),
|
|
"machine": platform.machine(),
|
|
"processor": platform.processor(),
|
|
"python_version": platform.python_version(),
|
|
},
|
|
"hardware": {
|
|
"cpu_count": psutil.cpu_count(),
|
|
"cpu_percent": psutil.cpu_percent(interval=1),
|
|
"memory": {
|
|
"total": psutil.virtual_memory().total,
|
|
"available": psutil.virtual_memory().available,
|
|
"percent": psutil.virtual_memory().percent,
|
|
"used": psutil.virtual_memory().used,
|
|
},
|
|
"disk": {
|
|
"total": psutil.disk_usage("/").total,
|
|
"used": psutil.disk_usage("/").used,
|
|
"free": psutil.disk_usage("/").free,
|
|
"percent": psutil.disk_usage("/").percent,
|
|
},
|
|
},
|
|
"network": {
|
|
"hostname": platform.node(),
|
|
"interfaces": _get_network_interfaces(),
|
|
},
|
|
"application": {
|
|
"version": "0.1.0",
|
|
"config_path": str(config.config_path),
|
|
"debug_mode": config.debug,
|
|
"uptime": str(datetime.now() - APP_START_TIME),
|
|
"server": {
|
|
"host": config.server.host,
|
|
"port": config.server.port,
|
|
"url": f"http://{config.server.host}:{config.server.port}",
|
|
"mdns_url": f"http://gsproapp.local:{config.server.port}" if config.server.mdns_enabled else None,
|
|
},
|
|
},
|
|
}
|
|
|
|
return system_info
|
|
|
|
|
|
@router.get("/gspro/status")
|
|
async def gspro_status():
|
|
"""
|
|
Get GSPro application status.
|
|
|
|
Returns information about the GSPro process if it's running.
|
|
"""
|
|
config = get_config()
|
|
|
|
is_running = is_gspro_running(config.gspro.window_title)
|
|
process_info = get_gspro_process_info() if is_running else None
|
|
|
|
return {
|
|
"running": is_running,
|
|
"window_title": config.gspro.window_title,
|
|
"process": process_info,
|
|
"auto_focus": config.gspro.auto_focus,
|
|
"key_delay": config.gspro.key_delay,
|
|
}
|
|
|
|
|
|
@router.get("/gspro/find")
|
|
async def find_gspro_window():
|
|
"""
|
|
Search for GSPro windows.
|
|
|
|
Helps users identify the correct window title for configuration.
|
|
"""
|
|
try:
|
|
import win32gui
|
|
|
|
def enum_window_callback(hwnd, windows):
|
|
if win32gui.IsWindowVisible(hwnd) and win32gui.IsWindowEnabled(hwnd):
|
|
window_text = win32gui.GetWindowText(hwnd)
|
|
if window_text and len(window_text) > 0:
|
|
# Look for windows that might be GSPro
|
|
if any(keyword in window_text.lower() for keyword in ["gspro", "golf", "simulator"]):
|
|
windows.append(
|
|
{"title": window_text, "hwnd": hwnd, "suggested": "gspro" in window_text.lower()}
|
|
)
|
|
return True
|
|
|
|
windows = []
|
|
win32gui.EnumWindows(enum_window_callback, windows)
|
|
|
|
return {
|
|
"found": len(windows) > 0,
|
|
"windows": windows,
|
|
"message": "Found potential GSPro windows" if windows else "No GSPro windows found",
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to search for GSPro windows: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.get("/metrics")
|
|
async def get_metrics():
|
|
"""
|
|
Get application metrics and performance statistics.
|
|
"""
|
|
config = get_config()
|
|
|
|
# Get current resource usage
|
|
process = psutil.Process()
|
|
|
|
metrics = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"uptime_seconds": int((datetime.now() - APP_START_TIME).total_seconds()),
|
|
"resources": {
|
|
"cpu_percent": process.cpu_percent(),
|
|
"memory_mb": process.memory_info().rss / 1024 / 1024,
|
|
"memory_percent": process.memory_percent(),
|
|
"threads": process.num_threads(),
|
|
"open_files": len(process.open_files()) if hasattr(process, "open_files") else 0,
|
|
"connections": len(process.connections()) if hasattr(process, "connections") else 0,
|
|
},
|
|
"system": {
|
|
"cpu_percent": psutil.cpu_percent(interval=0.5),
|
|
"memory_percent": psutil.virtual_memory().percent,
|
|
"disk_io": psutil.disk_io_counters()._asdict() if psutil.disk_io_counters() else {},
|
|
"network_io": psutil.net_io_counters()._asdict() if psutil.net_io_counters() else {},
|
|
},
|
|
}
|
|
|
|
return metrics
|
|
|
|
|
|
@router.get("/logs")
|
|
async def get_logs(lines: int = 100):
|
|
"""
|
|
Get recent application logs.
|
|
|
|
Args:
|
|
lines: Number of log lines to return (max 1000)
|
|
"""
|
|
# This is a placeholder - in production, you'd read from actual log files
|
|
return {
|
|
"message": "Log retrieval not yet implemented",
|
|
"lines_requested": min(lines, 1000),
|
|
"log_level": logging.getLevelName(logger.getEffectiveLevel()),
|
|
}
|
|
|
|
|
|
@router.post("/restart")
|
|
async def restart_application():
|
|
"""
|
|
Restart the application.
|
|
|
|
Note: This endpoint requires proper process management setup.
|
|
"""
|
|
# This would typically signal the process manager to restart
|
|
# For now, just return a message
|
|
return {
|
|
"success": False,
|
|
"message": "Application restart requires process manager setup. Please restart manually.",
|
|
}
|
|
|
|
|
|
@router.get("/dependencies")
|
|
async def check_dependencies():
|
|
"""
|
|
Check if all required dependencies are installed and accessible.
|
|
"""
|
|
dependencies = {
|
|
"required": {},
|
|
"optional": {},
|
|
}
|
|
|
|
# Check required dependencies
|
|
required_modules = [
|
|
"fastapi",
|
|
"uvicorn",
|
|
"pydantic",
|
|
"pydirectinput",
|
|
"mss",
|
|
"PIL",
|
|
"cv2",
|
|
"numpy",
|
|
"win32gui",
|
|
"psutil",
|
|
"zeroconf",
|
|
]
|
|
|
|
for module_name in required_modules:
|
|
try:
|
|
module = __import__(module_name)
|
|
version = getattr(module, "__version__", "unknown")
|
|
dependencies["required"][module_name] = {"installed": True, "version": version}
|
|
except ImportError:
|
|
dependencies["required"][module_name] = {"installed": False, "version": None}
|
|
|
|
# Check optional dependencies (V2 features)
|
|
optional_modules = [
|
|
"easyocr",
|
|
"pytesseract",
|
|
]
|
|
|
|
for module_name in optional_modules:
|
|
try:
|
|
module = __import__(module_name)
|
|
version = getattr(module, "__version__", "unknown")
|
|
dependencies["optional"][module_name] = {"installed": True, "version": version}
|
|
except ImportError:
|
|
dependencies["optional"][module_name] = {"installed": False, "version": None}
|
|
|
|
# Check if all required dependencies are installed
|
|
all_installed = all(dep["installed"] for dep in dependencies["required"].values())
|
|
|
|
return {
|
|
"all_required_installed": all_installed,
|
|
"dependencies": dependencies,
|
|
"message": "All required dependencies installed" if all_installed else "Some required dependencies are missing",
|
|
}
|
|
|
|
|
|
@router.get("/diagnostics")
|
|
async def run_diagnostics():
|
|
"""
|
|
Run comprehensive system diagnostics.
|
|
|
|
Checks various aspects of the system and application configuration.
|
|
"""
|
|
config = get_config()
|
|
diagnostics = {"timestamp": datetime.now().isoformat(), "checks": []}
|
|
|
|
# Check 1: GSPro connectivity
|
|
gspro_running = is_gspro_running(config.gspro.window_title)
|
|
diagnostics["checks"].append(
|
|
{
|
|
"name": "GSPro Connectivity",
|
|
"status": "pass" if gspro_running else "fail",
|
|
"message": "GSPro is running and accessible" if gspro_running else "GSPro window not found",
|
|
}
|
|
)
|
|
|
|
# Check 2: Network accessibility
|
|
try:
|
|
import socket
|
|
|
|
socket.create_connection(("8.8.8.8", 53), timeout=3)
|
|
network_ok = True
|
|
network_msg = "Network connection is working"
|
|
except:
|
|
network_ok = False
|
|
network_msg = "No internet connection detected"
|
|
|
|
diagnostics["checks"].append(
|
|
{
|
|
"name": "Network Connectivity",
|
|
"status": "pass" if network_ok else "warning",
|
|
"message": network_msg,
|
|
}
|
|
)
|
|
|
|
# Check 3: Configuration validity
|
|
config_valid = config.config_path and config.config_path.exists()
|
|
diagnostics["checks"].append(
|
|
{
|
|
"name": "Configuration File",
|
|
"status": "pass" if config_valid else "warning",
|
|
"message": f"Configuration file exists at {config.config_path}"
|
|
if config_valid
|
|
else "Configuration file not found",
|
|
}
|
|
)
|
|
|
|
# Check 4: Available disk space
|
|
disk_usage = psutil.disk_usage("/")
|
|
disk_ok = disk_usage.percent < 90
|
|
diagnostics["checks"].append(
|
|
{
|
|
"name": "Disk Space",
|
|
"status": "pass" if disk_ok else "warning",
|
|
"message": f"Disk usage at {disk_usage.percent:.1f}%"
|
|
+ (" - Consider freeing space" if not disk_ok else ""),
|
|
}
|
|
)
|
|
|
|
# Check 5: Memory availability
|
|
memory = psutil.virtual_memory()
|
|
memory_ok = memory.percent < 90
|
|
diagnostics["checks"].append(
|
|
{
|
|
"name": "Memory Availability",
|
|
"status": "pass" if memory_ok else "warning",
|
|
"message": f"Memory usage at {memory.percent:.1f}%"
|
|
+ (" - High memory usage detected" if not memory_ok else ""),
|
|
}
|
|
)
|
|
|
|
# Check 6: Python version
|
|
import sys
|
|
|
|
python_ok = sys.version_info >= (3, 11)
|
|
diagnostics["checks"].append(
|
|
{
|
|
"name": "Python Version",
|
|
"status": "pass" if python_ok else "warning",
|
|
"message": f"Python {platform.python_version()}"
|
|
+ (" - Consider upgrading to 3.11+" if not python_ok else ""),
|
|
}
|
|
)
|
|
|
|
# Calculate overall status
|
|
statuses = [check["status"] for check in diagnostics["checks"]]
|
|
if "fail" in statuses:
|
|
overall = "fail"
|
|
elif "warning" in statuses:
|
|
overall = "warning"
|
|
else:
|
|
overall = "pass"
|
|
|
|
diagnostics["overall_status"] = overall
|
|
diagnostics["summary"] = {
|
|
"passed": sum(1 for s in statuses if s == "pass"),
|
|
"warnings": sum(1 for s in statuses if s == "warning"),
|
|
"failures": sum(1 for s in statuses if s == "fail"),
|
|
}
|
|
|
|
return diagnostics
|
|
|
|
|
|
def _get_network_interfaces():
|
|
"""Helper function to get network interface information."""
|
|
interfaces = []
|
|
|
|
try:
|
|
for interface, addrs in psutil.net_if_addrs().items():
|
|
for addr in addrs:
|
|
if addr.family == socket.AF_INET: # IPv4
|
|
interfaces.append(
|
|
{
|
|
"name": interface,
|
|
"address": addr.address,
|
|
"netmask": addr.netmask,
|
|
"broadcast": addr.broadcast,
|
|
}
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Could not get network interfaces: {e}")
|
|
|
|
return interfaces
|