348 lines
10 KiB
Python
348 lines
10 KiB
Python
"""
|
|
Configuration API for managing GSPro Remote settings.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Any, Dict, Optional
|
|
|
|
from fastapi import APIRouter, BackgroundTasks, HTTPException, Query
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
from ..core.config import get_config, reset_config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter()
|
|
|
|
|
|
class ConfigUpdateRequest(BaseModel):
|
|
"""Request model for configuration updates."""
|
|
|
|
server: Optional[Dict[str, Any]] = Field(None, description="Server configuration")
|
|
capture: Optional[Dict[str, Any]] = Field(None, description="Capture configuration")
|
|
gspro: Optional[Dict[str, Any]] = Field(None, description="GSPro configuration")
|
|
vision: Optional[Dict[str, Any]] = Field(None, description="Vision configuration")
|
|
debug: Optional[bool] = Field(None, description="Debug mode")
|
|
|
|
|
|
class ConfigResponse(BaseModel):
|
|
"""Response model for configuration endpoints."""
|
|
|
|
success: bool
|
|
message: str
|
|
config: Optional[Dict[str, Any]] = None
|
|
|
|
|
|
class CaptureConfigUpdate(BaseModel):
|
|
"""Request model for capture configuration updates."""
|
|
|
|
fps: Optional[int] = Field(None, ge=1, le=60)
|
|
quality: Optional[int] = Field(None, ge=1, le=100)
|
|
resolution: Optional[str] = None
|
|
region_x: Optional[int] = Field(None, ge=0)
|
|
region_y: Optional[int] = Field(None, ge=0)
|
|
region_width: Optional[int] = Field(None, gt=0)
|
|
region_height: Optional[int] = Field(None, gt=0)
|
|
|
|
|
|
class ServerConfigUpdate(BaseModel):
|
|
"""Request model for server configuration updates."""
|
|
|
|
host: Optional[str] = None
|
|
port: Optional[int] = None
|
|
mdns_enabled: Optional[bool] = None
|
|
|
|
|
|
class GSProConfigUpdate(BaseModel):
|
|
"""Request model for GSPro configuration updates."""
|
|
|
|
window_title: Optional[str] = None
|
|
auto_focus: Optional[bool] = None
|
|
key_delay: Optional[float] = Field(None, ge=0, le=1)
|
|
|
|
|
|
class VisionConfigUpdate(BaseModel):
|
|
"""Request model for vision configuration updates."""
|
|
|
|
enabled: Optional[bool] = None
|
|
ocr_engine: Optional[str] = None
|
|
confidence_threshold: Optional[float] = Field(None, ge=0, le=1)
|
|
|
|
|
|
@router.get("/", response_model=Dict[str, Any])
|
|
async def get_configuration():
|
|
"""
|
|
Get the current application configuration.
|
|
|
|
Returns all configuration sections including server, capture, gspro, and vision settings.
|
|
"""
|
|
config = get_config()
|
|
return config.to_dict()
|
|
|
|
|
|
@router.put("/", response_model=ConfigResponse)
|
|
async def update_configuration(request: ConfigUpdateRequest):
|
|
"""
|
|
Update application configuration.
|
|
|
|
Only provided fields will be updated. Nested configuration objects are merged.
|
|
"""
|
|
config = get_config()
|
|
|
|
try:
|
|
update_dict = request.model_dump(exclude_none=True)
|
|
|
|
if not update_dict:
|
|
return ConfigResponse(success=False, message="No configuration changes provided")
|
|
|
|
# Update configuration
|
|
config.update(**update_dict)
|
|
|
|
logger.info(f"Configuration updated: {list(update_dict.keys())}")
|
|
|
|
return ConfigResponse(success=True, message="Configuration updated successfully", config=config.to_dict())
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to update configuration: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to update configuration: {str(e)}")
|
|
|
|
|
|
@router.post("/save", response_model=ConfigResponse)
|
|
async def save_configuration():
|
|
"""
|
|
Save the current configuration to disk.
|
|
|
|
This ensures changes persist across application restarts.
|
|
"""
|
|
config = get_config()
|
|
|
|
try:
|
|
config.save()
|
|
|
|
return ConfigResponse(
|
|
success=True, message=f"Configuration saved to {config.config_path}", config=config.to_dict()
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to save configuration: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to save configuration: {str(e)}")
|
|
|
|
|
|
@router.post("/reset", response_model=ConfigResponse)
|
|
async def reset_configuration():
|
|
"""
|
|
Reset configuration to default values.
|
|
|
|
This will overwrite any custom settings with the application defaults.
|
|
"""
|
|
config = get_config()
|
|
|
|
try:
|
|
config.reset()
|
|
|
|
logger.info("Configuration reset to defaults")
|
|
|
|
return ConfigResponse(success=True, message="Configuration reset to defaults", config=config.to_dict())
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to reset configuration: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to reset configuration: {str(e)}")
|
|
|
|
|
|
@router.get("/server")
|
|
async def get_server_config():
|
|
"""Get server-specific configuration."""
|
|
config = get_config()
|
|
return config.server.model_dump()
|
|
|
|
|
|
@router.put("/server")
|
|
async def update_server_config(request: ServerConfigUpdate):
|
|
"""
|
|
Update server configuration.
|
|
|
|
Note: Changing host or port requires application restart to take effect.
|
|
"""
|
|
config = get_config()
|
|
|
|
updates = request.model_dump(exclude_none=True)
|
|
|
|
if not updates:
|
|
return {"success": False, "message": "No updates provided"}
|
|
|
|
try:
|
|
config.update(server=updates)
|
|
|
|
return {
|
|
"success": True,
|
|
"message": "Server configuration updated (restart required for host/port changes)",
|
|
"config": config.server.model_dump(),
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to update server config: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.get("/capture")
|
|
async def get_capture_config():
|
|
"""Get capture/streaming configuration."""
|
|
config = get_config()
|
|
return config.capture.model_dump()
|
|
|
|
|
|
@router.put("/capture")
|
|
async def update_capture_config(request: CaptureConfigUpdate):
|
|
"""Update capture/streaming configuration."""
|
|
config = get_config()
|
|
|
|
updates = request.model_dump(exclude_none=True)
|
|
|
|
if not updates:
|
|
return {"success": False, "message": "No updates provided"}
|
|
|
|
try:
|
|
config.update(capture=updates)
|
|
|
|
return {"success": True, "message": "Capture configuration updated", "config": config.capture.model_dump()}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to update capture config: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.get("/gspro")
|
|
async def get_gspro_config():
|
|
"""Get GSPro-specific configuration."""
|
|
config = get_config()
|
|
return config.gspro.model_dump()
|
|
|
|
|
|
@router.put("/gspro")
|
|
async def update_gspro_config(request: GSProConfigUpdate):
|
|
"""Update GSPro-specific configuration."""
|
|
config = get_config()
|
|
|
|
updates = request.model_dump(exclude_none=True)
|
|
|
|
if not updates:
|
|
return {"success": False, "message": "No updates provided"}
|
|
|
|
try:
|
|
config.update(gspro=updates)
|
|
|
|
return {"success": True, "message": "GSPro configuration updated", "config": config.gspro.model_dump()}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to update GSPro config: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.get("/vision")
|
|
async def get_vision_config():
|
|
"""Get vision/OCR configuration (V2 features)."""
|
|
config = get_config()
|
|
return {**config.vision.model_dump(), "note": "Vision features are planned for V2"}
|
|
|
|
|
|
@router.put("/vision")
|
|
async def update_vision_config(request: VisionConfigUpdate):
|
|
"""
|
|
Update vision/OCR configuration.
|
|
|
|
Note: These features are planned for V2 and not yet implemented.
|
|
"""
|
|
config = get_config()
|
|
|
|
updates = request.model_dump(exclude_none=True)
|
|
|
|
if not updates:
|
|
return {"success": False, "message": "No updates provided"}
|
|
|
|
# Validate OCR engine if provided
|
|
if "ocr_engine" in updates and updates["ocr_engine"] not in ["easyocr", "tesseract"]:
|
|
raise HTTPException(status_code=400, detail="Invalid OCR engine. Must be 'easyocr' or 'tesseract'")
|
|
|
|
try:
|
|
config.update(vision=updates)
|
|
|
|
return {
|
|
"success": True,
|
|
"message": "Vision configuration updated (V2 features)",
|
|
"config": config.vision.model_dump(),
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to update vision config: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.post("/reload", response_model=ConfigResponse)
|
|
async def reload_configuration():
|
|
"""
|
|
Reload configuration from disk.
|
|
|
|
This discards any unsaved changes and reloads from the configuration file.
|
|
"""
|
|
try:
|
|
# Reset the global config instance
|
|
reset_config()
|
|
|
|
# Get new config (will load from disk)
|
|
config = get_config()
|
|
|
|
logger.info("Configuration reloaded from disk")
|
|
|
|
return ConfigResponse(
|
|
success=True, message=f"Configuration reloaded from {config.config_path}", config=config.to_dict()
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to reload configuration: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to reload configuration: {str(e)}")
|
|
|
|
|
|
@router.get("/path")
|
|
async def get_config_path():
|
|
"""Get the configuration file path."""
|
|
config = get_config()
|
|
return {"path": str(config.config_path), "exists": config.config_path.exists() if config.config_path else False}
|
|
|
|
|
|
@router.get("/validate")
|
|
async def validate_configuration():
|
|
"""
|
|
Validate the current configuration.
|
|
|
|
Checks for common issues and required settings.
|
|
"""
|
|
config = get_config()
|
|
|
|
issues = []
|
|
warnings = []
|
|
|
|
# Check server configuration
|
|
if config.server.port < 1024:
|
|
warnings.append("Server port is below 1024, may require administrator privileges")
|
|
|
|
# Check capture configuration
|
|
if config.capture.fps > 30:
|
|
warnings.append("High FPS may impact performance")
|
|
|
|
if config.capture.quality < 50:
|
|
warnings.append("Low JPEG quality may result in poor image clarity")
|
|
|
|
# Check GSPro configuration
|
|
if not config.gspro.window_title:
|
|
issues.append("GSPro window title is not set")
|
|
|
|
# Check vision configuration
|
|
if config.vision.enabled:
|
|
warnings.append("Vision features are enabled but not yet implemented (V2)")
|
|
|
|
return {
|
|
"valid": len(issues) == 0,
|
|
"issues": issues,
|
|
"warnings": warnings,
|
|
"summary": f"{len(issues)} issues, {len(warnings)} warnings",
|
|
}
|