rhill051/gsproremote
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Initial commit: GSPro Remote MVP - Phase 1 complete

Browse source
This commit is contained in:
Ryan Hill 2025-11-13 15:38:58 -06:00
commit 74ca4b38eb
50 changed files with 12818 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

167
backend/README.md Normal file
View file

@ -0,0 +1,167 @@
# GSPro Remote Backend
FastAPI-based backend service for GSPro Remote, providing keyboard control and screen streaming capabilities for GSPro golf simulator.
## Features
- **Keyboard Control API**: Send keyboard shortcuts to GSPro
- **Screen Streaming**: WebSocket-based map region streaming
- **Configuration Management**: Persistent settings storage
- **mDNS Discovery**: Auto-discoverable at `gsproapp.local`
- **Windows Integration**: Native Windows input simulation
## Requirements
- Python 3.11+
- Windows OS (for GSPro integration)
- GSPro running on the same machine
## Installation
### Using pip
```bash
python -m venv .venv
.venv\Scripts\activate
pip install -e .
```
### Using UV (recommended)
```bash
uv venv
uv pip install -e .
```
## Development Setup
1. Install development dependencies:
```bash
pip install -e ".[dev]"
```
2. Run the development server:
```bash
uvicorn app.main:app --reload --host 0.0.0.0 --port 5005
```
## API Structure
```
/api/actions/
POST /key - Send single key press
POST /keydown - Hold key down
POST /keyup - Release key
POST /combo - Send key combination
/api/config/
GET / - Get current configuration
PUT / - Update configuration
POST /save - Persist to disk
/api/vision/
WS /ws/stream - WebSocket map streaming
GET /regions - Get defined screen regions
POST /capture - Capture screen region
/api/system/
GET /health - Health check
GET /info - System information
```
## Configuration
Configuration is stored in `%LOCALAPPDATA%\GSPro Remote\config.json`
Default configuration:
```json
{
"server": {
"host": "0.0.0.0",
"port": 5005,
"mdns_enabled": true
},
"capture": {
"fps": 30,
"quality": 85,
"resolution": "720p"
},
"gspro": {
"window_title": "GSPro",
"auto_focus": true
}
}
```
## Project Structure
```
backend/
app/
__init__.py
main.py # FastAPI application
api/
actions.py # Keyboard control endpoints
config.py # Configuration endpoints
vision.py # Screen capture/streaming
system.py # System utilities
core/
config.py # Configuration management
input_ctrl.py # Windows input simulation
screen.py # Screen capture utilities
mdns.py # mDNS service registration
models/
requests.py # Pydantic request models
responses.py # Pydantic response models
tests/
test_*.py # Unit tests
pyproject.toml # Project dependencies
```
## Testing
Run tests with pytest:
```bash
pytest
```
With coverage:
```bash
pytest --cov=app --cov-report=html
```
## Building for Distribution
Build standalone executable:
```bash
pip install ".[build]"
python -m PyInstaller --onefile --name gspro-remote app/main.py
```
## Environment Variables
- `GSPRO_REMOTE_PORT`: Override default port (5005)
- `GSPRO_REMOTE_HOST`: Override default host (0.0.0.0)
- `GSPRO_REMOTE_CONFIG_PATH`: Override config location
- `GSPRO_REMOTE_DEBUG`: Enable debug logging
## Troubleshooting
### GSPro window not found
- Ensure GSPro is running
- Check window title matches configuration
- Run as administrator if permission issues
### Port already in use
- Check if another instance is running
- Change port in configuration
- Use `netstat -an | findstr :5005` to check
### mDNS not working
- Check Windows firewall settings
- Ensure Bonjour service is running
- Try accessing directly via IP instead
## License
MIT License - See parent LICENSE file for details

13
backend/app/__init__.py Normal file
View file

@ -0,0 +1,13 @@
"""
GSPro Remote Backend Application
A FastAPI-based backend service for controlling GSPro golf simulator
via keyboard shortcuts and providing screen streaming capabilities.
"""
__version__ = "0.1.0"
__author__ = "GSPro Remote Team"
from .main import app
__all__ = ["app", "__version__"]

7
backend/app/api/__init__.py Normal file
View file

@ -0,0 +1,7 @@
"""
API module for GSPro Remote backend.
"""
from . import actions, config, system, vision
__all__ = ["actions", "config", "system", "vision"]

232
backend/app/api/actions.py Normal file
View file

@ -0,0 +1,232 @@
"""
Actions API for sending keyboard inputs to GSPro.
"""
import asyncio
import logging
from typing import Optional
from fastapi import APIRouter, HTTPException, BackgroundTasks
from pydantic import BaseModel, Field
from ..core.input_ctrl import press_key, press_keys, key_down, key_up, focus_window, is_gspro_running
from ..core.config import get_config
logger = logging.getLogger(__name__)
router = APIRouter()
class KeyPressRequest(BaseModel):
"""Request model for single key press."""
key: str = Field(..., description="Key or key combination to press (e.g., 'a', 'ctrl+m')")
delay: Optional[float] = Field(0.0, description="Delay in seconds before pressing the key")
class KeyHoldRequest(BaseModel):
"""Request model for key hold operations."""
key: str = Field(..., description="Key to hold or release")
duration: Optional[float] = Field(None, description="Duration to hold the key (seconds)")
class KeySequenceRequest(BaseModel):
"""Request model for key sequence."""
keys: list[str] = Field(..., description="List of keys to press in sequence")
interval: float = Field(0.1, description="Interval between key presses (seconds)")
class ActionResponse(BaseModel):
"""Response model for action endpoints."""
success: bool
message: str
key: Optional[str] = None
@router.post("/key", response_model=ActionResponse)
async def send_key(request: KeyPressRequest):
"""
Send a single key press or key combination to GSPro.
Supports:
- Single keys: 'a', 'space', 'enter'
- Combinations: 'ctrl+m', 'shift+tab'
- Function keys: 'f1', 'f11'
- Arrow keys: 'up', 'down', 'left', 'right'
"""
config = get_config()
# Check if GSPro is running
if not is_gspro_running():
raise HTTPException(status_code=409, detail="GSPro is not running or window not found")
# Focus GSPro window if auto-focus is enabled
if config.gspro.auto_focus:
if not focus_window(config.gspro.window_title):
logger.warning(f"Could not focus window: {config.gspro.window_title}")
try:
# Apply delay if specified
if request.delay > 0:
await asyncio.sleep(request.delay)
# Send the key press
if "+" in request.key:
# Handle key combination
press_keys(request.key)
else:
# Handle single key
press_key(request.key)
logger.info(f"Sent key press: {request.key}")
return ActionResponse(success=True, message=f"Key '{request.key}' pressed successfully", key=request.key)
except Exception as e:
logger.error(f"Failed to send key press: {e}")
raise HTTPException(status_code=500, detail=f"Failed to send key press: {str(e)}")
@router.post("/keydown", response_model=ActionResponse)
async def send_key_down(request: KeyHoldRequest, background_tasks: BackgroundTasks):
"""
Press and hold a key down.
If duration is specified, the key will be automatically released after that time.
Otherwise, you must call /keyup to release it.
"""
config = get_config()
if not is_gspro_running():
raise HTTPException(status_code=409, detail="GSPro is not running or window not found")
if config.gspro.auto_focus:
focus_window(config.gspro.window_title)
try:
key_down(request.key)
logger.info(f"Key down: {request.key}")
# If duration is specified, schedule key release
if request.duration:
async def release_key():
await asyncio.sleep(request.duration)
key_up(request.key)
logger.info(f"Key released after {request.duration}s: {request.key}")
background_tasks.add_task(release_key)
return ActionResponse(
success=True, message=f"Key '{request.key}' held for {request.duration}s", key=request.key
)
else:
return ActionResponse(
success=True, message=f"Key '{request.key}' pressed down (call /keyup to release)", key=request.key
)
except Exception as e:
logger.error(f"Failed to hold key down: {e}")
raise HTTPException(status_code=500, detail=f"Failed to hold key down: {str(e)}")
@router.post("/keyup", response_model=ActionResponse)
async def send_key_up(request: KeyHoldRequest):
"""
Release a held key.
"""
config = get_config()
if not is_gspro_running():
raise HTTPException(status_code=409, detail="GSPro is not running or window not found")
if config.gspro.auto_focus:
focus_window(config.gspro.window_title)
try:
key_up(request.key)
logger.info(f"Key up: {request.key}")
return ActionResponse(success=True, message=f"Key '{request.key}' released", key=request.key)
except Exception as e:
logger.error(f"Failed to release key: {e}")
raise HTTPException(status_code=500, detail=f"Failed to release key: {str(e)}")
@router.post("/sequence", response_model=ActionResponse)
async def send_key_sequence(request: KeySequenceRequest):
"""
Send a sequence of key presses with specified interval between them.
"""
config = get_config()
if not is_gspro_running():
raise HTTPException(status_code=409, detail="GSPro is not running or window not found")
if config.gspro.auto_focus:
focus_window(config.gspro.window_title)
try:
for i, key in enumerate(request.keys):
if "+" in key:
press_keys(key)
else:
press_key(key)
# Add interval between keys (except after last one)
if i < len(request.keys) - 1:
await asyncio.sleep(request.interval)
logger.info(f"Sent key sequence: {request.keys}")
return ActionResponse(
success=True, message=f"Sent {len(request.keys)} key presses", key=", ".join(request.keys)
)
except Exception as e:
logger.error(f"Failed to send key sequence: {e}")
raise HTTPException(status_code=500, detail=f"Failed to send key sequence: {str(e)}")
@router.get("/shortcuts")
async def get_shortcuts():
"""
Get a list of all available GSPro keyboard shortcuts.
"""
shortcuts = {
"aim": {"up": "up", "down": "down", "left": "left", "right": "right", "reset": "a"},
"club": {"up": "u", "down": "k"},
"shot": {"mulligan": "ctrl+m", "options": "'", "putt_toggle": "u"},
"tee": {"left": "c", "right": "v"},
"view": {
"map_toggle": "s",
"map_zoom_in": "q",
"map_zoom_out": "w",
"scorecard": "t",
"range_finder": "r",
"heat_map": "y",
"pin_indicator": "p",
"flyover": "o",
"free_look": "f5",
"aim_point": "f3",
"green_grid": "g",
"ui_toggle": "h",
},
"camera": {"go_to_ball": "5", "fly_to_ball": "6", "shot_camera": "j"},
"practice": {"go_to_ball": "8", "previous_hole": "9", "next_hole": "0"},
"system": {
"fullscreen": "f11",
"fps_display": "f",
"console_short": "f8",
"console_tall": "f9",
"tracer_clear": "f1",
},
"settings": {
"sound_on": "+",
"sound_off": "-",
"lighting": "l",
"3d_grass": "z",
"switch_hand": "n",
"shadow_increase": ">",
"shadow_decrease": "<",
},
}
return {"shortcuts": shortcuts, "total": sum(len(category) for category in shortcuts.values())}

348
backend/app/api/config.py Normal file
View file

@ -0,0 +1,348 @@
"""
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",
}

409
backend/app/api/system.py Normal file
View file

@ -0,0 +1,409 @@
"""
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

345
backend/app/api/vision.py Normal file
View file

@ -0,0 +1,345 @@
"""
Vision API for screen capture and streaming.
Currently implements WebSocket streaming for the map panel.
OCR and advanced vision features are gated behind configuration flags for V2.
"""
import asyncio
import json
import logging
from typing import Optional, Dict, Any
from fastapi import APIRouter, WebSocket, WebSocketDisconnect, HTTPException, Depends
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from ..core.config import get_config
from ..core.screen import get_screen_capture, capture_region
logger = logging.getLogger(__name__)
router = APIRouter()
class StreamConfig(BaseModel):
"""Configuration for video streaming."""
fps: int = Field(30, ge=1, le=60, description="Frames per second")
quality: int = Field(85, ge=1, le=100, description="JPEG quality")
resolution: str = Field("720p", description="Stream resolution preset")
region_x: int = Field(0, ge=0, description="Region X coordinate")
region_y: int = Field(0, ge=0, description="Region Y coordinate")
region_width: int = Field(640, ge=1, description="Region width")
region_height: int = Field(480, ge=1, description="Region height")
class RegionDefinition(BaseModel):
"""Definition of a screen region for capture."""
name: str = Field(..., description="Region name")
x: int = Field(..., ge=0, description="X coordinate")
y: int = Field(..., ge=0, description="Y coordinate")
width: int = Field(..., gt=0, description="Width")
height: int = Field(..., gt=0, description="Height")
description: Optional[str] = Field(None, description="Region description")
class CaptureRequest(BaseModel):
"""Request to capture a screen region."""
x: int = Field(0, ge=0, description="X coordinate")
y: int = Field(0, ge=0, description="Y coordinate")
width: int = Field(640, gt=0, description="Width")
height: int = Field(480, gt=0, description="Height")
format: str = Field("base64", description="Output format (base64 or raw)")
quality: int = Field(85, ge=1, le=100, description="JPEG quality")
class StreamManager:
"""Manages WebSocket streaming sessions."""
def __init__(self):
self.active_streams: Dict[str, WebSocket] = {}
self.stream_configs: Dict[str, StreamConfig] = {}
async def add_stream(self, client_id: str, websocket: WebSocket, config: StreamConfig):
"""Add a new streaming session."""
await websocket.accept()
self.active_streams[client_id] = websocket
self.stream_configs[client_id] = config
logger.info(f"Stream started for client {client_id}")
async def remove_stream(self, client_id: str):
"""Remove a streaming session."""
if client_id in self.active_streams:
del self.active_streams[client_id]
del self.stream_configs[client_id]
logger.info(f"Stream stopped for client {client_id}")
async def stream_frame(self, client_id: str, frame_data: str):
"""Send a frame to a specific client."""
if client_id in self.active_streams:
websocket = self.active_streams[client_id]
try:
await websocket.send_text(frame_data)
except Exception as e:
logger.error(f"Failed to send frame to client {client_id}: {e}")
await self.remove_stream(client_id)
# Global stream manager
stream_manager = StreamManager()
@router.websocket("/ws/stream")
async def stream_video(websocket: WebSocket):
"""
WebSocket endpoint for streaming video of a screen region.
The client should send a JSON message with stream configuration:
{
"action": "start",
"config": {
"fps": 30,
"quality": 85,
"resolution": "720p",
"region_x": 0,
"region_y": 0,
"region_width": 640,
"region_height": 480
}
}
"""
config = get_config()
client_id = id(websocket)
try:
# Accept the WebSocket connection
await websocket.accept()
logger.info(f"WebSocket connected: client {client_id}")
# Wait for initial configuration message
data = await websocket.receive_text()
message = json.loads(data)
if message.get("action") != "start":
await websocket.send_json({"error": "First message must be start action"})
await websocket.close()
return
# Parse stream configuration
stream_config = StreamConfig(**message.get("config", {}))
# Override with server config if needed
stream_config.fps = min(stream_config.fps, config.capture.fps)
stream_config.quality = config.capture.quality
# Add to stream manager
await stream_manager.add_stream(str(client_id), websocket, stream_config)
# Send confirmation
await websocket.send_json({"type": "config", "data": stream_config.model_dump()})
# Get screen capture instance
capture = get_screen_capture()
# Calculate frame interval
frame_interval = 1.0 / stream_config.fps
# Streaming loop
while True:
try:
# Capture the region
image = capture.capture_region(
stream_config.region_x,
stream_config.region_y,
stream_config.region_width,
stream_config.region_height,
)
# Resize if needed
if stream_config.resolution != "native":
target_width, target_height = capture.get_resolution_preset(stream_config.resolution)
# Only resize if different from capture size
if target_width != stream_config.region_width or target_height != stream_config.region_height:
image = capture.resize_image(image, width=target_width)
# Convert to base64
base64_image = capture.image_to_base64(image, quality=stream_config.quality)
# Send frame
frame_data = json.dumps(
{"type": "frame", "data": base64_image, "timestamp": asyncio.get_event_loop().time()}
)
await websocket.send_text(frame_data)
# Wait for next frame
await asyncio.sleep(frame_interval)
except WebSocketDisconnect:
logger.info(f"Client {client_id} disconnected")
break
except Exception as e:
logger.error(f"Error in stream loop for client {client_id}: {e}")
await websocket.send_json({"type": "error", "message": str(e)})
break
except Exception as e:
logger.error(f"WebSocket error for client {client_id}: {e}")
finally:
await stream_manager.remove_stream(str(client_id))
logger.info(f"WebSocket closed: client {client_id}")
@router.post("/capture", response_model=Dict[str, Any])
async def capture_screen_region(request: CaptureRequest):
"""
Capture a single frame from a screen region.
This is useful for testing or getting a single snapshot.
"""
try:
capture = get_screen_capture()
# Capture the region
image = capture.capture_region(request.x, request.y, request.width, request.height)
if request.format == "base64":
# Convert to base64
base64_image = capture.image_to_base64(image, quality=request.quality)
return {
"success": True,
"format": "base64",
"data": base64_image,
"width": request.width,
"height": request.height,
}
else:
return {"success": False, "error": f"Unsupported format: {request.format}"}
except Exception as e:
logger.error(f"Failed to capture region: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.get("/regions")
async def get_regions():
"""
Get predefined screen regions for capture.
Returns common regions like map area, club indicator, etc.
"""
config = get_config()
# Predefined regions for GSPro UI elements
regions = {
"map": {
"name": "Map Panel",
"x": config.capture.region_x,
"y": config.capture.region_y,
"width": config.capture.region_width,
"height": config.capture.region_height,
"description": "GSPro mini-map or expanded map view",
},
"club": {
"name": "Club Indicator",
"x": 50,
"y": 200,
"width": 200,
"height": 100,
"description": "Current club selection display",
},
"shot_info": {
"name": "Shot Information",
"x": 50,
"y": 50,
"width": 300,
"height": 150,
"description": "Shot distance and trajectory information",
},
"scorecard": {
"name": "Scorecard",
"x": 400,
"y": 100,
"width": 800,
"height": 600,
"description": "Scorecard overlay when visible",
},
}
return {"regions": regions, "total": len(regions)}
@router.post("/regions/{region_name}")
async def update_region(region_name: str, region: RegionDefinition):
"""
Update or create a screen region definition.
This allows users to define custom regions for their setup.
"""
config = get_config()
if region_name == "map":
# Update the map region in config
config.capture.region_x = region.x
config.capture.region_y = region.y
config.capture.region_width = region.width
config.capture.region_height = region.height
config.save()
return {"success": True, "message": f"Region '{region_name}' updated", "region": region.model_dump()}
else:
# For now, only map region is persisted
# V2 will add support for custom regions
return {"success": False, "message": "Custom regions not yet supported (V2 feature)"}
# OCR endpoints - gated behind vision config flag
@router.post("/ocr")
async def perform_ocr(request: CaptureRequest):
"""
Perform OCR on a screen region (V2 feature).
This endpoint is only available when vision features are enabled.
"""
config = get_config()
if not config.vision.enabled:
raise HTTPException(status_code=403, detail="Vision features are not enabled. This is a V2 feature.")
# OCR implementation will go here in V2
return {"success": False, "message": "OCR features coming in V2", "vision_enabled": config.vision.enabled}
@router.get("/markers")
async def get_markers():
"""
Get visual markers for template matching (V2 feature).
This endpoint is only available when vision features are enabled.
"""
config = get_config()
if not config.vision.enabled:
raise HTTPException(status_code=403, detail="Vision features are not enabled. This is a V2 feature.")
# Marker management will go here in V2
return {"markers": [], "message": "Marker features coming in V2", "vision_enabled": config.vision.enabled}
@router.get("/status")
async def get_vision_status():
"""Get the status of vision features."""
config = get_config()
return {
"streaming_enabled": True,
"ocr_enabled": config.vision.enabled,
"markers_enabled": config.vision.enabled,
"active_streams": len(stream_manager.active_streams),
"capture_config": {
"fps": config.capture.fps,
"quality": config.capture.quality,
"resolution": config.capture.resolution,
},
}

21
backend/app/core/__init__.py Normal file
View file

@ -0,0 +1,21 @@
"""
Core modules for GSPro Remote backend.
"""
from .config import AppConfig, get_config
from .input_ctrl import press_key, press_keys, key_down, key_up, focus_window, is_gspro_running
from .screen import capture_screen, get_screen_size, capture_region
__all__ = [
"AppConfig",
"get_config",
"press_key",
"press_keys",
"key_down",
"key_up",
"focus_window",
"is_gspro_running",
"capture_screen",
"get_screen_size",
"capture_region",
]

193
backend/app/core/config.py Normal file
View file

@ -0,0 +1,193 @@
"""
Configuration management for GSPro Remote.
"""
import json
import logging
from pathlib import Path
from typing import Optional
from functools import lru_cache
from pydantic import BaseModel, Field
from pydantic_settings import BaseSettings
logger = logging.getLogger(__name__)
class ServerConfig(BaseModel):
"""Server configuration settings."""
host: str = Field("0.0.0.0", description="Server host address")
port: int = Field(5005, description="Server port")
mdns_enabled: bool = Field(True, description="Enable mDNS service discovery")
class CaptureConfig(BaseModel):
"""Screen capture configuration settings."""
fps: int = Field(30, description="Frames per second for streaming")
quality: int = Field(85, description="JPEG quality (0-100)")
resolution: str = Field("720p", description="Stream resolution")
region_x: int = Field(0, description="Map region X coordinate")
region_y: int = Field(0, description="Map region Y coordinate")
region_width: int = Field(640, description="Map region width")
region_height: int = Field(480, description="Map region height")
class GSProConfig(BaseModel):
"""GSPro application configuration settings."""
window_title: str = Field("GSPro", description="GSPro window title")
auto_focus: bool = Field(True, description="Auto-focus GSPro window before sending keys")
key_delay: float = Field(0.05, description="Default delay between key presses (seconds)")
class VisionConfig(BaseModel):
"""Computer vision configuration settings (for V2 features)."""
enabled: bool = Field(False, description="Enable vision features")
ocr_engine: str = Field("easyocr", description="OCR engine to use (easyocr or tesseract)")
confidence_threshold: float = Field(0.7, description="Minimum confidence for OCR detection")
class AppConfig(BaseSettings):
"""Main application configuration."""
server: ServerConfig = Field(default_factory=ServerConfig)
capture: CaptureConfig = Field(default_factory=CaptureConfig)
gspro: GSProConfig = Field(default_factory=GSProConfig)
vision: VisionConfig = Field(default_factory=VisionConfig)
config_path: Optional[Path] = None
debug: bool = Field(False, description="Enable debug mode")
class Config:
env_prefix = "GSPRO_REMOTE_"
env_nested_delimiter = "__"
case_sensitive = False
def __init__(self, **kwargs):
super().__init__(**kwargs)
if self.config_path is None:
self.config_path = self._get_default_config_path()
self.load()
@staticmethod
def _get_default_config_path() -> Path:
"""Get the default configuration file path."""
import os
if os.name == "nt": # Windows
base_path = Path(os.environ.get("LOCALAPPDATA", ""))
if not base_path:
base_path = Path.home() / "AppData" / "Local"
else: # Unix-like
base_path = Path.home() / ".config"
config_dir = base_path / "GSPro Remote"
config_dir.mkdir(parents=True, exist_ok=True)
return config_dir / "config.json"
def load(self) -> None:
"""Load configuration from file."""
if self.config_path and self.config_path.exists():
try:
with open(self.config_path, "r") as f:
data = json.load(f)
# Update configuration with loaded data
if "server" in data:
self.server = ServerConfig(**data["server"])
if "capture" in data:
self.capture = CaptureConfig(**data["capture"])
if "gspro" in data:
self.gspro = GSProConfig(**data["gspro"])
if "vision" in data:
self.vision = VisionConfig(**data["vision"])
if "debug" in data:
self.debug = data["debug"]
logger.info(f"Configuration loaded from {self.config_path}")
except Exception as e:
logger.warning(f"Failed to load configuration: {e}")
self.save() # Save default configuration
else:
# Create default configuration file
self.save()
logger.info(f"Created default configuration at {self.config_path}")
def save(self) -> None:
"""Save configuration to file."""
if self.config_path:
try:
self.config_path.parent.mkdir(parents=True, exist_ok=True)
data = {
"server": self.server.model_dump(),
"capture": self.capture.model_dump(),
"gspro": self.gspro.model_dump(),
"vision": self.vision.model_dump(),
"debug": self.debug,
}
with open(self.config_path, "w") as f:
json.dump(data, f, indent=2)
logger.info(f"Configuration saved to {self.config_path}")
except Exception as e:
logger.error(f"Failed to save configuration: {e}")
def update(self, **kwargs) -> None:
"""Update configuration with new values."""
for key, value in kwargs.items():
if hasattr(self, key):
if isinstance(value, dict):
# Update nested configuration
current = getattr(self, key)
if isinstance(current, BaseModel):
for sub_key, sub_value in value.items():
if hasattr(current, sub_key):
setattr(current, sub_key, sub_value)
else:
setattr(self, key, value)
self.save()
def reset(self) -> None:
"""Reset configuration to defaults."""
self.server = ServerConfig()
self.capture = CaptureConfig()
self.gspro = GSProConfig()
self.vision = VisionConfig()
self.debug = False
self.save()
def to_dict(self) -> dict:
"""Convert configuration to dictionary."""
return {
"server": self.server.model_dump(),
"capture": self.capture.model_dump(),
"gspro": self.gspro.model_dump(),
"vision": self.vision.model_dump(),
"debug": self.debug,
"config_path": str(self.config_path) if self.config_path else None,
}
# Global configuration instance
_config: Optional[AppConfig] = None
@lru_cache(maxsize=1)
def get_config() -> AppConfig:
"""Get the global configuration instance."""
global _config
if _config is None:
_config = AppConfig()
return _config
def reset_config() -> None:
"""Reset the global configuration instance."""
global _config
_config = None
get_config.cache_clear()

350
backend/app/core/input_ctrl.py Normal file
View file

@ -0,0 +1,350 @@
"""
Windows input control module for simulating keyboard inputs to GSPro.
"""
import logging
import time
from typing import Optional, List
try:
import pydirectinput
import win32gui
import win32con
import win32process
import psutil
except ImportError as e:
raise ImportError(f"Required Windows dependencies not installed: {e}")
logger = logging.getLogger(__name__)
# Configure pydirectinput
pydirectinput.PAUSE = 0.01 # Reduce default pause between actions
def is_gspro_running(window_title: str = "GSPro") -> bool:
"""
Check if GSPro is running by looking for its window.
Args:
window_title: The window title to search for
Returns:
True if GSPro window is found, False otherwise
"""
def enum_window_callback(hwnd, windows):
if win32gui.IsWindowVisible(hwnd) and win32gui.IsWindowEnabled(hwnd):
window_text = win32gui.GetWindowText(hwnd)
if window_title.lower() in window_text.lower():
windows.append(hwnd)
return True
windows = []
win32gui.EnumWindows(enum_window_callback, windows)
return len(windows) > 0
def find_gspro_window(window_title: str = "GSPro") -> Optional[int]:
"""
Find the GSPro window handle.
Args:
window_title: The window title to search for
Returns:
Window handle if found, None otherwise
"""
def enum_window_callback(hwnd, result):
window_text = win32gui.GetWindowText(hwnd)
if window_title.lower() in window_text.lower():
result.append(hwnd)
return True
result = []
win32gui.EnumWindows(enum_window_callback, result)
if result:
return result[0]
return None
def focus_window(window_title: str = "GSPro") -> bool:
"""
Focus the GSPro window to ensure it receives keyboard input.
Args:
window_title: The window title to focus
Returns:
True if window was focused successfully, False otherwise
"""
try:
hwnd = find_gspro_window(window_title)
if hwnd:
# Restore window if minimized
if win32gui.IsIconic(hwnd):
win32gui.ShowWindow(hwnd, win32con.SW_RESTORE)
# Set foreground window
win32gui.SetForegroundWindow(hwnd)
# Small delay to ensure window is focused
time.sleep(0.1)
logger.debug(f"Focused window: {window_title}")
return True
else:
logger.warning(f"Window not found: {window_title}")
return False
except Exception as e:
logger.error(f"Failed to focus window: {e}")
return False
def press_key(key: str, interval: float = 0.0) -> None:
"""
Simulate a single key press.
Args:
key: The key to press (e.g., 'a', 'space', 'f1', 'up')
interval: Time to wait after pressing the key
"""
try:
# Normalize key name for pydirectinput
key_normalized = key.lower().strip()
# Handle special key mappings
key_mappings = {
"ctrl": "ctrl",
"control": "ctrl",
"alt": "alt",
"shift": "shift",
"tab": "tab",
"space": "space",
"enter": "enter",
"return": "enter",
"escape": "esc",
"esc": "esc",
"backspace": "backspace",
"delete": "delete",
"del": "delete",
"insert": "insert",
"ins": "insert",
"home": "home",
"end": "end",
"pageup": "pageup",
"pagedown": "pagedown",
"up": "up",
"down": "down",
"left": "left",
"right": "right",
"plus": "+",
"minus": "-",
"apostrophe": "'",
"quote": "'",
}
# Map key if needed
key_to_press = key_mappings.get(key_normalized, key_normalized)
# Press the key
pydirectinput.press(key_to_press)
if interval > 0:
time.sleep(interval)
logger.debug(f"Pressed key: {key}")
except Exception as e:
logger.error(f"Failed to press key '{key}': {e}")
raise
def press_keys(keys: str, interval: float = 0.0) -> None:
"""
Simulate a key combination or sequence.
Args:
keys: Key combination string (e.g., 'ctrl+m', 'shift+tab')
interval: Time to wait after pressing the keys
"""
try:
# Check if it's a key combination
if "+" in keys:
# Split into modifiers and key
parts = keys.lower().split("+")
modifiers = []
main_key = parts[-1]
# Identify modifiers
for part in parts[:-1]:
if part in ["ctrl", "control"]:
modifiers.append("ctrl")
elif part in ["alt"]:
modifiers.append("alt")
elif part in ["shift"]:
modifiers.append("shift")
elif part in ["win", "windows", "cmd", "command"]:
modifiers.append("win")
# Press combination using hotkey
if modifiers:
hotkey_parts = modifiers + [main_key]
pydirectinput.hotkey(*hotkey_parts)
else:
press_key(main_key)
else:
# Single key press
press_key(keys)
if interval > 0:
time.sleep(interval)
logger.debug(f"Pressed keys: {keys}")
except Exception as e:
logger.error(f"Failed to press keys '{keys}': {e}")
raise
def key_down(key: str) -> None:
"""
Hold a key down.
Args:
key: The key to hold down
"""
try:
key_normalized = key.lower().strip()
pydirectinput.keyDown(key_normalized)
logger.debug(f"Key down: {key}")
except Exception as e:
logger.error(f"Failed to hold key down '{key}': {e}")
raise
def key_up(key: str) -> None:
"""
Release a held key.
Args:
key: The key to release
"""
try:
key_normalized = key.lower().strip()
pydirectinput.keyUp(key_normalized)
logger.debug(f"Key up: {key}")
except Exception as e:
logger.error(f"Failed to release key '{key}': {e}")
raise
def type_text(text: str, interval: float = 0.0) -> None:
"""
Type a string of text.
Args:
text: The text to type
interval: Time between each character
"""
try:
pydirectinput.typewrite(text, interval=interval)
logger.debug(f"Typed text: {text[:20]}...")
except Exception as e:
logger.error(f"Failed to type text: {e}")
raise
def mouse_click(x: Optional[int] = None, y: Optional[int] = None, button: str = "left") -> None:
"""
Simulate a mouse click.
Args:
x: X coordinate (None for current position)
y: Y coordinate (None for current position)
button: Mouse button ('left', 'right', 'middle')
"""
try:
if x is not None and y is not None:
pydirectinput.click(x, y, button=button)
logger.debug(f"Mouse click at ({x}, {y}) with {button} button")
else:
pydirectinput.click(button=button)
logger.debug(f"Mouse click with {button} button")
except Exception as e:
logger.error(f"Failed to perform mouse click: {e}")
raise
def mouse_move(x: int, y: int, duration: float = 0.0) -> None:
"""
Move the mouse cursor.
Args:
x: Target X coordinate
y: Target Y coordinate
duration: Time to take for the movement
"""
try:
if duration > 0:
pydirectinput.moveTo(x, y, duration=duration)
else:
pydirectinput.moveTo(x, y)
logger.debug(f"Mouse moved to ({x}, {y})")
except Exception as e:
logger.error(f"Failed to move mouse: {e}")
raise
def get_gspro_process_info() -> Optional[dict]:
"""
Get information about the GSPro process if it's running.
Returns:
Dictionary with process info or None if not found
"""
try:
for proc in psutil.process_iter(["pid", "name", "cpu_percent", "memory_info"]):
if "gspro" in proc.info["name"].lower():
return {
"pid": proc.info["pid"],
"name": proc.info["name"],
"cpu_percent": proc.info["cpu_percent"],
"memory_mb": proc.info["memory_info"].rss / 1024 / 1024 if proc.info["memory_info"] else 0,
}
except Exception as e:
logger.error(f"Failed to get GSPro process info: {e}")
return None
# Test function for development
def test_input_control():
"""Test function to verify input control is working."""
print("Testing input control...")
# Check if GSPro is running
if is_gspro_running():
print("✓ GSPro is running")
# Try to focus the window
if focus_window():
print("✓ GSPro window focused")
else:
print("✗ Could not focus GSPro window")
else:
print("✗ GSPro is not running")
print("Please start GSPro and try again")
return
# Get process info
info = get_gspro_process_info()
if info:
print(
f"✓ GSPro process found: PID={info['pid']}, CPU={info['cpu_percent']:.1f}%, Memory={info['memory_mb']:.1f}MB"
)
print("\nInput control test complete!")
if __name__ == "__main__":
# Run test when module is executed directly
test_input_control()

335
backend/app/core/mdns.py Normal file
View file

@ -0,0 +1,335 @@
"""
mDNS service registration for GSPro Remote.
Allows the application to be discovered on the local network.
"""
import logging
import socket
import threading
from typing import Optional, Dict, Any
try:
from zeroconf import ServiceInfo, Zeroconf, IPVersion
except ImportError as e:
raise ImportError(f"Zeroconf library not installed: {e}")
logger = logging.getLogger(__name__)
class MDNSService:
"""
Manages mDNS/Bonjour service registration for network discovery.
"""
def __init__(
self,
name: str = "gsproapp",
port: int = 5005,
service_type: str = "_http._tcp.local.",
properties: Optional[Dict[str, Any]] = None,
):
"""
Initialize mDNS service.
Args:
name: Service name (will be accessible as {name}.local)
port: Port number the service is running on
service_type: mDNS service type
properties: Additional service properties
"""
self.name = name
self.port = port
self.service_type = service_type
self.properties = properties or {}
self.zeroconf: Optional[Zeroconf] = None
self.service_info: Optional[ServiceInfo] = None
self.is_running = False
self._lock = threading.Lock()
def _get_local_ip(self) -> str:
"""
Get the local IP address of the machine.
Returns:
Local IP address as string
"""
try:
# Create a socket to determine the local IP
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
# Connect to a public DNS server to determine local interface
s.connect(("8.8.8.8", 80))
return s.getsockname()[0]
except Exception:
# Fallback to localhost if can't determine
return "127.0.0.1"
def _create_service_info(self) -> ServiceInfo:
"""
Create the ServiceInfo object for registration.
Returns:
Configured ServiceInfo object
"""
local_ip = self._get_local_ip()
hostname = socket.gethostname()
# Create fully qualified service name
service_name = f"{self.name}.{self.service_type}"
# Add default properties
default_properties = {
"version": "0.1.0",
"platform": "windows",
"api": "rest",
"ui": "web",
}
# Merge with custom properties
all_properties = {**default_properties, **self.properties}
# Convert properties to bytes
properties_bytes = {}
for key, value in all_properties.items():
if isinstance(value, str):
properties_bytes[key] = value.encode("utf-8")
elif isinstance(value, bytes):
properties_bytes[key] = value
else:
properties_bytes[key] = str(value).encode("utf-8")
# Create service info
service_info = ServiceInfo(
type_=self.service_type,
name=service_name,
addresses=[socket.inet_aton(local_ip)],
port=self.port,
properties=properties_bytes,
server=f"{hostname}.local.",
)
return service_info
def start(self) -> bool:
"""
Start the mDNS service registration.
Returns:
True if service started successfully, False otherwise
"""
with self._lock:
if self.is_running:
logger.warning("mDNS service is already running")
return True
try:
# Create Zeroconf instance
self.zeroconf = Zeroconf(ip_version=IPVersion.V4Only)
# Create and register service
self.service_info = self._create_service_info()
self.zeroconf.register_service(self.service_info)
self.is_running = True
logger.info(f"mDNS service registered: {self.name}.local:{self.port} (type: {self.service_type})")
return True
except Exception as e:
logger.error(f"Failed to start mDNS service: {e}")
self.cleanup()
return False
def stop(self) -> None:
"""Stop the mDNS service registration."""
with self._lock:
if not self.is_running:
return
self.cleanup()
self.is_running = False
logger.info("mDNS service stopped")
def cleanup(self) -> None:
"""Clean up mDNS resources."""
try:
if self.zeroconf and self.service_info:
self.zeroconf.unregister_service(self.service_info)
if self.zeroconf:
self.zeroconf.close()
self.zeroconf = None
self.service_info = None
except Exception as e:
logger.error(f"Error during mDNS cleanup: {e}")
def update_properties(self, properties: Dict[str, Any]) -> bool:
"""
Update service properties.
Args:
properties: New properties to set
Returns:
True if properties updated successfully, False otherwise
"""
with self._lock:
if not self.is_running:
logger.warning("Cannot update properties: service is not running")
return False
try:
self.properties.update(properties)
# Recreate and re-register service with new properties
if self.zeroconf and self.service_info:
self.zeroconf.unregister_service(self.service_info)
self.service_info = self._create_service_info()
self.zeroconf.register_service(self.service_info)
logger.info("mDNS service properties updated")
return True
except Exception as e:
logger.error(f"Failed to update mDNS properties: {e}")
return False
def get_url(self) -> str:
"""
Get the URL for accessing the service.
Returns:
Service URL
"""
return f"http://{self.name}.local:{self.port}"
def __enter__(self):
"""Context manager entry."""
self.start()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Context manager exit."""
self.stop()
class MDNSBrowser:
"""
Browse for mDNS services on the network.
Useful for discovering other GSPro Remote instances.
"""
def __init__(self, service_type: str = "_http._tcp.local."):
"""
Initialize mDNS browser.
Args:
service_type: Type of services to browse for
"""
self.service_type = service_type
self.services: Dict[str, Dict[str, Any]] = {}
self.zeroconf: Optional[Zeroconf] = None
def browse(self, timeout: float = 5.0) -> Dict[str, Dict[str, Any]]:
"""
Browse for services on the network.
Args:
timeout: Time to wait for services (seconds)
Returns:
Dictionary of discovered services
"""
try:
from zeroconf import ServiceBrowser, ServiceListener
import time
class Listener(ServiceListener):
def __init__(self, browser):
self.browser = browser
def add_service(self, zeroconf, service_type, name):
info = zeroconf.get_service_info(service_type, name)
if info:
self.browser.services[name] = {
"name": name,
"address": socket.inet_ntoa(info.addresses[0]) if info.addresses else None,
"port": info.port,
"properties": info.properties,
}
def remove_service(self, zeroconf, service_type, name):
self.browser.services.pop(name, None)
def update_service(self, zeroconf, service_type, name):
pass
self.zeroconf = Zeroconf(ip_version=IPVersion.V4Only)
listener = Listener(self)
browser = ServiceBrowser(self.zeroconf, self.service_type, listener)
# Wait for services to be discovered
time.sleep(timeout)
browser.cancel()
self.zeroconf.close()
return self.services
except Exception as e:
logger.error(f"Failed to browse for services: {e}")
return {}
# Test function for development
def test_mdns_service():
"""Test mDNS service registration."""
import time
print("Testing mDNS service registration...")
# Test service registration
service = MDNSService(
name="gsproapp-test",
port=5005,
properties={"test": "true", "instance": "development"},
)
if service.start():
print(f"✓ mDNS service started: {service.get_url()}")
print(f" You should be able to access it at: http://gsproapp-test.local:5005")
# Keep service running for 10 seconds
print(" Service will run for 10 seconds...")
time.sleep(10)
# Test property update
if service.update_properties({"status": "running", "uptime": "10s"}):
print("✓ Properties updated successfully")
service.stop()
print("✓ mDNS service stopped")
else:
print("✗ Failed to start mDNS service")
# Test service browsing
print("\nBrowsing for HTTP services on the network...")
browser = MDNSBrowser()
services = browser.browse(timeout=3.0)
if services:
print(f"Found {len(services)} services:")
for name, info in services.items():
print(f" - {name}: {info['address']}:{info['port']}")
else:
print("No services found")
print("\nmDNS test complete!")
if __name__ == "__main__":
test_mdns_service()

370
backend/app/core/screen.py Normal file
View file

@ -0,0 +1,370 @@
"""
Screen capture utilities for GSPro Remote.
"""
import logging
from typing import Optional, Tuple, Dict, Any
from io import BytesIO
import base64
try:
import mss
import mss.tools
from PIL import Image
import cv2
import numpy as np
except ImportError as e:
raise ImportError(f"Required screen capture dependencies not installed: {e}")
logger = logging.getLogger(__name__)
class ScreenCapture:
"""Manages screen capture operations."""
def __init__(self):
"""Initialize screen capture manager."""
self.sct = mss.mss()
self._monitor_info = None
def get_monitors(self) -> list[dict]:
"""
Get information about all available monitors.
Returns:
List of monitor information dictionaries
"""
monitors = []
for i, monitor in enumerate(self.sct.monitors):
monitors.append(
{
"index": i,
"left": monitor["left"],
"top": monitor["top"],
"width": monitor["width"],
"height": monitor["height"],
"is_primary": i == 0, # Index 0 is combined virtual screen
}
)
return monitors
def get_primary_monitor(self) -> dict:
"""
Get the primary monitor information.
Returns:
Primary monitor information
"""
# Index 1 is typically the primary monitor in mss
return self.sct.monitors[1] if len(self.sct.monitors) > 1 else self.sct.monitors[0]
def capture_screen(self, monitor_index: int = 1) -> np.ndarray:
"""
Capture the entire screen.
Args:
monitor_index: Index of the monitor to capture (0 for all, 1 for primary)
Returns:
Captured screen as numpy array (BGR format)
"""
try:
monitor = self.sct.monitors[monitor_index]
screenshot = self.sct.grab(monitor)
# Convert to numpy array (BGR format for OpenCV compatibility)
img = np.array(screenshot)
img = cv2.cvtColor(img, cv2.COLOR_RGBA2BGR)
return img
except Exception as e:
logger.error(f"Failed to capture screen: {e}")
raise
def capture_region(self, x: int, y: int, width: int, height: int, monitor_index: int = 1) -> np.ndarray:
"""
Capture a specific region of the screen.
Args:
x: X coordinate of the region (relative to monitor)
y: Y coordinate of the region (relative to monitor)
width: Width of the region
height: Height of the region
monitor_index: Index of the monitor to capture from
Returns:
Captured region as numpy array (BGR format)
"""
try:
monitor = self.sct.monitors[monitor_index]
# Define region to capture
region = {
"left": monitor["left"] + x,
"top": monitor["top"] + y,
"width": width,
"height": height,
}
screenshot = self.sct.grab(region)
# Convert to numpy array
img = np.array(screenshot)
img = cv2.cvtColor(img, cv2.COLOR_RGBA2BGR)
return img
except Exception as e:
logger.error(f"Failed to capture region: {e}")
raise
def capture_window(self, window_title: str) -> Optional[np.ndarray]:
"""
Capture a specific window by title.
Args:
window_title: Title of the window to capture
Returns:
Captured window as numpy array or None if window not found
"""
try:
import win32gui
def enum_window_callback(hwnd, windows):
if win32gui.IsWindowVisible(hwnd) and win32gui.IsWindowEnabled(hwnd):
window_text = win32gui.GetWindowText(hwnd)
if window_title.lower() in window_text.lower():
windows.append(hwnd)
return True
windows = []
win32gui.EnumWindows(enum_window_callback, windows)
if not windows:
logger.warning(f"Window not found: {window_title}")
return None
# Get window rectangle
hwnd = windows[0]
rect = win32gui.GetWindowRect(hwnd)
x, y, x2, y2 = rect
width = x2 - x
height = y2 - y
# Capture the window region
return self.capture_region(x, y, width, height, monitor_index=0)
except Exception as e:
logger.error(f"Failed to capture window: {e}")
return None
def image_to_base64(self, image: np.ndarray, quality: int = 85, format: str = "JPEG") -> str:
"""
Convert an image array to base64 string.
Args:
image: Image as numpy array (BGR format)
quality: JPEG quality (1-100)
format: Image format (JPEG, PNG)
Returns:
Base64 encoded image string
"""
try:
# Convert BGR to RGB for PIL
rgb_image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
pil_image = Image.fromarray(rgb_image)
# Save to bytes
buffer = BytesIO()
if format.upper() == "JPEG":
pil_image.save(buffer, format=format, quality=quality, optimize=True)
else:
pil_image.save(buffer, format=format)
# Encode to base64
buffer.seek(0)
base64_string = base64.b64encode(buffer.getvalue()).decode("utf-8")
return base64_string
except Exception as e:
logger.error(f"Failed to convert image to base64: {e}")
raise
def resize_image(self, image: np.ndarray, width: Optional[int] = None, height: Optional[int] = None) -> np.ndarray:
"""
Resize an image while maintaining aspect ratio.
Args:
image: Image as numpy array
width: Target width (None to calculate from height)
height: Target height (None to calculate from width)
Returns:
Resized image as numpy array
"""
try:
h, w = image.shape[:2]
if width and not height:
# Calculate height maintaining aspect ratio
height = int(h * (width / w))
elif height and not width:
# Calculate width maintaining aspect ratio
width = int(w * (height / h))
elif not width and not height:
# No resize needed
return image
return cv2.resize(image, (width, height), interpolation=cv2.INTER_AREA)
except Exception as e:
logger.error(f"Failed to resize image: {e}")
raise
def get_resolution_preset(self, preset: str) -> Tuple[int, int]:
"""
Get width and height for a resolution preset.
Args:
preset: Resolution preset (e.g., '720p', '1080p', '480p')
Returns:
Tuple of (width, height)
"""
presets = {
"480p": (854, 480),
"540p": (960, 540),
"720p": (1280, 720),
"900p": (1600, 900),
"1080p": (1920, 1080),
"1440p": (2560, 1440),
"2160p": (3840, 2160),
"4k": (3840, 2160),
}
return presets.get(preset.lower(), (1280, 720))
def close(self):
"""Close the screen capture resources."""
if hasattr(self, "sct"):
self.sct.close()
# Global screen capture instance
_screen_capture: Optional[ScreenCapture] = None
def get_screen_capture() -> ScreenCapture:
"""
Get the global screen capture instance.
Returns:
ScreenCapture instance
"""
global _screen_capture
if _screen_capture is None:
_screen_capture = ScreenCapture()
return _screen_capture
def capture_screen(monitor_index: int = 1) -> np.ndarray:
"""
Capture the entire screen.
Args:
monitor_index: Index of the monitor to capture
Returns:
Captured screen as numpy array
"""
return get_screen_capture().capture_screen(monitor_index)
def capture_region(x: int, y: int, width: int, height: int) -> np.ndarray:
"""
Capture a specific region of the screen.
Args:
x: X coordinate of the region
y: Y coordinate of the region
width: Width of the region
height: Height of the region
Returns:
Captured region as numpy array
"""
return get_screen_capture().capture_region(x, y, width, height)
def get_screen_size() -> Tuple[int, int]:
"""
Get the primary screen size.
Returns:
Tuple of (width, height)
"""
monitor = get_screen_capture().get_primary_monitor()
return monitor["width"], monitor["height"]
def capture_gspro_window(window_title: str = "GSPro") -> Optional[np.ndarray]:
"""
Capture the GSPro window.
Args:
window_title: GSPro window title
Returns:
Captured window as numpy array or None if not found
"""
return get_screen_capture().capture_window(window_title)
# Test function for development
def test_screen_capture():
"""Test screen capture functionality."""
print("Testing screen capture...")
capture = ScreenCapture()
# Get monitor information
monitors = capture.get_monitors()
print(f"Found {len(monitors)} monitors:")
for monitor in monitors:
print(
f" Monitor {monitor['index']}: {monitor['width']}x{monitor['height']} at ({monitor['left']}, {monitor['top']})"
)
# Capture primary screen
try:
screen = capture.capture_screen()
print(f"✓ Captured primary screen: {screen.shape}")
except Exception as e:
print(f"✗ Failed to capture screen: {e}")
# Test region capture
try:
region = capture.capture_region(100, 100, 640, 480)
print(f"✓ Captured region: {region.shape}")
except Exception as e:
print(f"✗ Failed to capture region: {e}")
# Test image to base64 conversion
try:
base64_str = capture.image_to_base64(region)
print(f"✓ Converted to base64: {len(base64_str)} chars")
except Exception as e:
print(f"✗ Failed to convert to base64: {e}")
# Test resolution presets
presets = ["480p", "720p", "1080p"]
for preset in presets:
width, height = capture.get_resolution_preset(preset)
print(f" {preset}: {width}x{height}")
capture.close()
print("\nScreen capture test complete!")
if __name__ == "__main__":
test_screen_capture()

115
backend/app/main.py Normal file
View file

@ -0,0 +1,115 @@
"""
Main FastAPI application for GSPro Remote backend.
"""
import logging
from contextlib import asynccontextmanager
from pathlib import Path
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from .api import actions, config, system, vision
from .core.config import AppConfig, get_config
from .core.mdns import MDNSService
# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)
@asynccontextmanager
async def lifespan(app: FastAPI):
"""
Application lifespan manager for startup/shutdown tasks.
"""
# Startup
logger.info("Starting GSPro Remote backend v0.1.0")
# Load configuration
config = get_config()
logger.info(f"Configuration loaded from {config.config_path}")
# Start mDNS service if enabled
mdns_service = None
if config.server.mdns_enabled:
try:
mdns_service = MDNSService(name="gsproapp", port=config.server.port, service_type="_http._tcp.local.")
mdns_service.start()
logger.info(f"mDNS service started: gsproapp.local:{config.server.port}")
except Exception as e:
logger.warning(f"Failed to start mDNS service: {e}")
yield
# Shutdown
logger.info("Shutting down GSPro Remote backend")
# Stop mDNS service
if mdns_service:
mdns_service.stop()
logger.info("mDNS service stopped")
# Save configuration
config.save()
logger.info("Configuration saved")
def create_app() -> FastAPI:
"""
Create and configure the FastAPI application.
"""
app = FastAPI(
title="GSPro Remote",
version="0.1.0",
description="Remote control API for GSPro golf simulator",
docs_url="/api/docs",
redoc_url="/api/redoc",
openapi_url="/api/openapi.json",
lifespan=lifespan,
)
# Configure CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # In production, specify actual origins
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Mount API routers
app.include_router(actions.router, prefix="/api/actions", tags=["Actions"])
app.include_router(config.router, prefix="/api/config", tags=["Configuration"])
app.include_router(vision.router, prefix="/api/vision", tags=["Vision"])
app.include_router(system.router, prefix="/api/system", tags=["System"])
# Serve frontend UI if built
ui_path = Path(__file__).parent.parent / "ui"
if ui_path.exists():
app.mount("/ui", StaticFiles(directory=str(ui_path), html=True), name="ui")
logger.info(f"Serving UI from {ui_path}")
# Root redirect
@app.get("/")
async def root():
return {
"name": "GSPro Remote",
"version": "0.1.0",
"status": "running",
"ui": "/ui" if ui_path.exists() else None,
"docs": "/api/docs",
}
return app
# Create the application instance
app = create_app()
if __name__ == "__main__":
import uvicorn
config = get_config()
uvicorn.run("app.main:app", host=config.server.host, port=config.server.port, reload=True, log_level="info")

76
backend/pyproject.toml Normal file
View file

@ -0,0 +1,76 @@
[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "gspro-remote"
version = "0.1.0"
description = "Remote control application for GSPro golf simulator"
readme = "README.md"
requires-python = ">=3.11"
license = {text = "MIT"}
authors = [
{name = "GSPro Remote Team"},
]
dependencies = [
"fastapi>=0.104.0",
"uvicorn[standard]>=0.24.0",
"pydantic>=2.5.0",
"pydantic-settings>=2.1.0",
"pydirectinput>=1.0.4",
"pywin32>=306",
"mss>=9.0.1",
"opencv-python>=4.8.0",
"pillow>=10.1.0",
"zeroconf>=0.120.0",
"websockets>=12.0",
"python-multipart>=0.0.6",
"aiofiles>=23.2.1",
]
[project.optional-dependencies]
dev = [
"pytest>=7.4.0",
"pytest-asyncio>=0.21.0",
"pytest-cov>=4.1.0",
"black>=23.11.0",
"ruff>=0.1.0",
"mypy>=1.7.0",
"httpx>=0.25.0",
]
vision = [
"easyocr>=1.7.0",
"pytesseract>=0.3.10",
"numpy>=1.24.0",
]
build = [
"pyinstaller>=6.0.0",
"nuitka>=1.8.0",
]
[tool.setuptools.packages.find]
where = ["."]
include = ["app*"]
exclude = ["tests*"]
[tool.black]
line-length = 120
target-version = ['py311']
[tool.ruff]
line-length = 120
select = ["E", "F", "I", "N", "W"]
ignore = ["E501"]
target-version = "py311"
[tool.mypy]
python_version = "3.11"
warn_return_any = true
warn_unused_configs = true
ignore_missing_imports = true
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py"]
python_functions = ["test_*"]
addopts = "-v --cov=app --cov-report=html --cov-report=term"

27
backend/requirements.txt Normal file
View file

@ -0,0 +1,27 @@
# Core dependencies
fastapi>=0.104.0
uvicorn[standard]>=0.24.0
pydantic>=2.5.0
pydantic-settings>=2.1.0
# Windows input control
pydirectinput>=1.0.4
pywin32>=306; sys_platform == 'win32'
# Screen capture and image processing
mss>=9.0.1
opencv-python>=4.8.0
pillow>=10.1.0
# Networking and service discovery
zeroconf>=0.120.0
websockets>=12.0
python-multipart>=0.0.6
aiofiles>=23.2.1
# System utilities
psutil>=5.9.0
numpy>=1.24.0
# HTTP client for testing
httpx>=0.25.0