media-player-server/media_server/routes/display.py

"""Display brightness, power, contrast, input-source, color-preset and picture-mode API."""

import asyncio
import logging

from fastapi import APIRouter, Depends
from pydantic import BaseModel, Field

from ..auth import verify_token
from ..services.display_service import (
    list_monitors,
    set_brightness,
    set_color_preset,
    set_contrast,
    set_input_source,
    set_picture_mode,
    set_power,
)

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api/display", tags=["display"])


class BrightnessRequest(BaseModel):
    brightness: int = Field(ge=0, le=100)


class PowerRequest(BaseModel):
    on: bool


class ContrastRequest(BaseModel):
    contrast: int = Field(ge=0, le=100)


class InputSourceRequest(BaseModel):
    source: str


class ColorPresetRequest(BaseModel):
    preset: str


class PictureModeRequest(BaseModel):
    code: int = Field(ge=0, le=255)


# DDC/CI hardware writes open a per-monitor handle and can take seconds —
# every public endpoint dispatches into a worker thread so the event loop
# stays responsive.


@router.get("/monitors")
async def get_monitors(
    refresh: bool = False,
    rediscover: bool = False,
    _: str = Depends(verify_token),
) -> list[dict]:
    """List all connected monitors with their reported DDC/CI capabilities."""
    monitors = await asyncio.to_thread(
        list_monitors, force_refresh=refresh, rediscover=rediscover
    )
    logger.debug("Found %d monitors", len(monitors))
    return [m.to_dict() for m in monitors]


@router.post("/brightness/{monitor_id}")
async def set_monitor_brightness(
    monitor_id: int, request: BrightnessRequest, _: str = Depends(verify_token)
) -> dict:
    """Set brightness for a specific monitor."""
    success = await asyncio.to_thread(set_brightness, monitor_id, request.brightness)
    if success:
        logger.info("Set monitor %d brightness to %d", monitor_id, request.brightness)
    return {"success": success}


@router.post("/power/{monitor_id}")
async def set_monitor_power(
    monitor_id: int, request: PowerRequest, _: str = Depends(verify_token)
) -> dict:
    """Turn a monitor on or off."""
    action = "on" if request.on else "off"
    success = await asyncio.to_thread(set_power, monitor_id, request.on)
    if success:
        logger.info("Set monitor %d power %s", monitor_id, action)
    return {"success": success}


@router.post("/contrast/{monitor_id}")
async def set_monitor_contrast(
    monitor_id: int, request: ContrastRequest, _: str = Depends(verify_token)
) -> dict:
    """Set DDC/CI contrast for a specific monitor."""
    success = await asyncio.to_thread(set_contrast, monitor_id, request.contrast)
    if success:
        logger.info("Set monitor %d contrast to %d", monitor_id, request.contrast)
    return {"success": success}


@router.post("/input_source/{monitor_id}")
async def set_monitor_input_source(
    monitor_id: int, request: InputSourceRequest, _: str = Depends(verify_token)
) -> dict:
    """Switch a monitor's DDC/CI input source (e.g. HDMI1, DP1)."""
    success = await asyncio.to_thread(set_input_source, monitor_id, request.source)
    if success:
        logger.info("Set monitor %d input source to %s", monitor_id, request.source)
    return {"success": success}


@router.post("/color_preset/{monitor_id}")
async def set_monitor_color_preset(
    monitor_id: int, request: ColorPresetRequest, _: str = Depends(verify_token)
) -> dict:
    """Apply a DDC/CI color preset (color temperature) to the monitor."""
    success = await asyncio.to_thread(set_color_preset, monitor_id, request.preset)
    if success:
        logger.info("Set monitor %d color preset to %s", monitor_id, request.preset)
    return {"success": success}


@router.post("/picture_mode/{monitor_id}")
async def set_monitor_picture_mode(
    monitor_id: int, request: PictureModeRequest, _: str = Depends(verify_token)
) -> dict:
    """Apply a DDC/CI picture/scene mode (VCP 0xDC) by raw code."""
    success = await asyncio.to_thread(set_picture_mode, monitor_id, request.code)
    if success:
        logger.info("Set monitor %d picture mode to code %d", monitor_id, request.code)
    return {"success": success}