"""Media control API endpoints.""" import logging from fastapi import APIRouter, Depends, HTTPException, Query, WebSocket, WebSocketDisconnect from fastapi import status from fastapi.responses import Response from ..auth import verify_token, verify_token_or_query from ..config import settings from ..models import MediaStatus, VolumeRequest, SeekRequest from ..services import get_media_controller, get_current_album_art from ..services.websocket_manager import ws_manager logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/media", tags=["media"]) @router.get("/status", response_model=MediaStatus) async def get_media_status(_: str = Depends(verify_token)) -> MediaStatus: """Get current media playback status. Returns: Current playback state, media info, volume, etc. """ controller = get_media_controller() return await controller.get_status() @router.post("/play") async def play(_: str = Depends(verify_token)) -> dict: """Resume or start playback. Returns: Success status """ controller = get_media_controller() success = await controller.play() if not success: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Failed to start playback - no active media session", ) return {"success": True} @router.post("/pause") async def pause(_: str = Depends(verify_token)) -> dict: """Pause playback. Returns: Success status """ controller = get_media_controller() success = await controller.pause() if not success: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Failed to pause - no active media session", ) return {"success": True} @router.post("/stop") async def stop(_: str = Depends(verify_token)) -> dict: """Stop playback. Returns: Success status """ controller = get_media_controller() success = await controller.stop() if not success: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Failed to stop - no active media session", ) return {"success": True} @router.post("/next") async def next_track(_: str = Depends(verify_token)) -> dict: """Skip to next track. Returns: Success status """ controller = get_media_controller() success = await controller.next_track() if not success: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Failed to skip - no active media session", ) return {"success": True} @router.post("/previous") async def previous_track(_: str = Depends(verify_token)) -> dict: """Go to previous track. Returns: Success status """ controller = get_media_controller() success = await controller.previous_track() if not success: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Failed to go back - no active media session", ) return {"success": True} @router.post("/volume") async def set_volume( request: VolumeRequest, _: str = Depends(verify_token) ) -> dict: """Set the system volume. Args: request: Volume level (0-100) Returns: Success status with new volume level """ controller = get_media_controller() success = await controller.set_volume(request.volume) if not success: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Failed to set volume", ) return {"success": True, "volume": request.volume} @router.post("/mute") async def toggle_mute(_: str = Depends(verify_token)) -> dict: """Toggle mute state. Returns: Success status with new mute state """ controller = get_media_controller() muted = await controller.toggle_mute() return {"success": True, "muted": muted} @router.post("/seek") async def seek(request: SeekRequest, _: str = Depends(verify_token)) -> dict: """Seek to a position in the current track. Args: request: Position in seconds Returns: Success status """ controller = get_media_controller() success = await controller.seek(request.position) if not success: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Failed to seek - no active media session or seek not supported", ) return {"success": True, "position": request.position} @router.get("/artwork") async def get_artwork(_: str = Depends(verify_token_or_query)) -> Response: """Get the current album artwork. Returns: The album art image as PNG/JPEG """ art_bytes = get_current_album_art() if art_bytes is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="No album artwork available", ) # Try to detect image type from magic bytes content_type = "image/png" # Default if art_bytes[:3] == b"\xff\xd8\xff": content_type = "image/jpeg" elif art_bytes[:8] == b"\x89PNG\r\n\x1a\n": content_type = "image/png" elif art_bytes[:4] == b"RIFF" and art_bytes[8:12] == b"WEBP": content_type = "image/webp" return Response(content=art_bytes, media_type=content_type) @router.websocket("/ws") async def websocket_endpoint( websocket: WebSocket, token: str = Query(..., description="API authentication token"), ) -> None: """WebSocket endpoint for real-time media status updates. Authentication is done via query parameter since WebSocket doesn't support custom headers in the browser. Messages sent to client: - {"type": "status", "data": {...}} - Initial status on connect - {"type": "status_update", "data": {...}} - Status changes - {"type": "error", "message": "..."} - Error messages Client can send: - {"type": "ping"} - Keepalive, server responds with {"type": "pong"} - {"type": "get_status"} - Request current status """ # Verify token if token != settings.api_token: await websocket.close(code=4001, reason="Invalid authentication token") return await ws_manager.connect(websocket) try: while True: # Wait for messages from client (for keepalive/ping) data = await websocket.receive_json() if data.get("type") == "ping": await websocket.send_json({"type": "pong"}) elif data.get("type") == "get_status": # Allow manual status request controller = get_media_controller() status_data = await controller.get_status() await websocket.send_json({ "type": "status", "data": status_data.model_dump(), }) except WebSocketDisconnect: await ws_manager.disconnect(websocket) except Exception as e: logger.error("WebSocket error: %s", e) await ws_manager.disconnect(websocket)