Add ADB-based Android screen capture engine with display picker integration

New scrcpy/ADB capture engine that captures Android device screens over
ADB using screencap polling. Supports USB and WiFi ADB connections with
device auto-discovery. Engine-aware display picker shows Android devices
when scrcpy engine is selected, with inline ADB connect form for WiFi
devices.

Key changes:
- New scrcpy_engine.py using adb screencap polling (~1-2 FPS over WiFi)
- Engine-aware GET /config/displays?engine_type= API
- ADB connect/disconnect API endpoints (POST /adb/connect, /adb/disconnect)
- Display picker supports engine-specific device lists
- Stream/test modals pass engine type to display picker
- Test template handler changed to sync def to prevent event loop blocking
- Restart script merges registry PATH for newly-installed tools
- All engines (including unavailable) shown in engine list with status flag

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-23 18:06:15 +03:00
parent cc08bb1c19
commit 199039326b
12 changed files with 644 additions and 26 deletions

View File

@@ -1,10 +1,13 @@
"""System routes: health, version, displays, performance."""
"""System routes: health, version, displays, performance, ADB."""
import subprocess
import sys
from datetime import datetime
from typing import Optional
import psutil
from fastapi import APIRouter, HTTPException
from fastapi import APIRouter, HTTPException, Query
from pydantic import BaseModel
from wled_controller import __version__
from wled_controller.api.auth import AuthRequired
@@ -73,16 +76,26 @@ async def get_version():
@router.get("/api/v1/config/displays", response_model=DisplayListResponse, tags=["Config"])
async def get_displays(_: AuthRequired):
async def get_displays(
_: AuthRequired,
engine_type: Optional[str] = Query(None, description="Engine type to get displays for"),
):
"""Get list of available displays.
Returns information about all available monitors/displays that can be captured.
When ``engine_type`` is provided, returns displays specific to that engine
(e.g. ``scrcpy`` returns connected Android devices instead of desktop monitors).
"""
logger.info("Listing available displays")
logger.info(f"Listing available displays (engine_type={engine_type})")
try:
# Get available displays with all metadata (name, refresh rate, etc.)
display_dataclasses = get_available_displays()
if engine_type:
from wled_controller.core.capture_engines import EngineRegistry
engine_cls = EngineRegistry.get_engine(engine_type)
display_dataclasses = engine_cls.get_available_displays()
else:
display_dataclasses = get_available_displays()
# Convert dataclass DisplayInfo to Pydantic DisplayInfo
displays = [
@@ -106,6 +119,8 @@ async def get_displays(_: AuthRequired):
count=len(displays),
)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"Failed to get displays: {e}")
raise HTTPException(
@@ -175,3 +190,70 @@ def get_system_performance(_: AuthRequired):
gpu=gpu,
timestamp=datetime.utcnow(),
)
# ---------------------------------------------------------------------------
# ADB helpers (for Android / scrcpy engine)
# ---------------------------------------------------------------------------
class AdbConnectRequest(BaseModel):
address: str
def _get_adb_path() -> str:
"""Get the adb binary path from the scrcpy engine's resolver."""
from wled_controller.core.capture_engines.scrcpy_engine import _get_adb
return _get_adb()
@router.post("/api/v1/adb/connect", tags=["ADB"])
async def adb_connect(_: AuthRequired, request: AdbConnectRequest):
"""Connect to a WiFi ADB device by IP address.
Appends ``:5555`` if no port is specified.
"""
address = request.address.strip()
if not address:
raise HTTPException(status_code=400, detail="Address is required")
if ":" not in address:
address = f"{address}:5555"
adb = _get_adb_path()
logger.info(f"Connecting ADB device: {address}")
try:
result = subprocess.run(
[adb, "connect", address],
capture_output=True, text=True, timeout=10,
)
output = (result.stdout + result.stderr).strip()
if "connected" in output.lower():
return {"status": "connected", "address": address, "message": output}
raise HTTPException(status_code=400, detail=output or "Connection failed")
except FileNotFoundError:
raise HTTPException(
status_code=500,
detail="adb not found on PATH. Install Android SDK Platform-Tools.",
)
except subprocess.TimeoutExpired:
raise HTTPException(status_code=504, detail="ADB connect timed out")
@router.post("/api/v1/adb/disconnect", tags=["ADB"])
async def adb_disconnect(_: AuthRequired, request: AdbConnectRequest):
"""Disconnect a WiFi ADB device."""
address = request.address.strip()
if not address:
raise HTTPException(status_code=400, detail="Address is required")
adb = _get_adb_path()
logger.info(f"Disconnecting ADB device: {address}")
try:
result = subprocess.run(
[adb, "disconnect", address],
capture_output=True, text=True, timeout=10,
)
return {"status": "disconnected", "message": result.stdout.strip()}
except FileNotFoundError:
raise HTTPException(status_code=500, detail="adb not found on PATH")
except subprocess.TimeoutExpired:
raise HTTPException(status_code=504, detail="ADB disconnect timed out")