api/server.py

#!/usr/bin/env python3
"""
API server for Synesthesia runtime.
Provides REST and WebSocket endpoints for controlling the runtime and accessing metrics.
"""

import logging
import asyncio
import json
import os
import sys
from typing import Dict, Any, Optional
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException
from fastapi.middleware.cors import CORSMiddleware
import uvicorn

# Add the project root to the sys.path so we can import the runtime module
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))

from ML_Pipeline.shared.env import apply_defaults

# Import runtime components (we'll import them as they become available)
try:
    from runtime.camera_capture import CameraCapture
    from runtime.mic_capture import MicCapture
    from runtime.gemma_prompt_engine import GemmaPromptEngine
    from runtime.magenta_generation import MagentaGeneration
    from runtime.clip_scheduler import ClipScheduler
except ImportError as e:
    logging.warning(f"Some runtime modules not available: {e}")
    # We'll create placeholder classes for now
    CameraCapture = MicCapture = GemmaPromptEngine = MagentaGeneration = ClipScheduler = None

logger = logging.getLogger(__name__)

app = FastAPI(title="Synesthesia Runtime API", version="0.1.0")

# Apply canonical defaults from environment/dotenv
env_config = apply_defaults()

# CORS configuration
allowed_origins = [o.strip() for o in env_config.get("ALLOWED_ORIGINS", "http://localhost:1420").split(",")]
allow_credentials = env_config.get("ALLOW_CREDENTIALS", "True").lower() == "true"

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=allowed_origins,
    allow_credentials=allow_credentials,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Global runtime components
camera: Optional[CameraCapture] = None
mic: Optional[MicCapture] = None
prompt_engine: Optional[GemmaPromptEngine] = None
music_generator: Optional[MagentaGeneration] = None
clip_scheduler: Optional[ClipScheduler] = None

# WebSocket connections for real-time updates
class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            try:
                await connection.send_text(message)
            except:
                # Remove broken connections
                self.active_connections.remove(connection)

manager = ConnectionManager()

@app.on_event("startup")
async def startup_event():
    """Initialize runtime components on startup."""
    global camera, mic, prompt_engine, music_generator, clip_scheduler
    logger.info("Starting Synesthesia runtime components...")
    
    try:
        # Initialize camera
        camera = CameraCapture()
        if camera.start():
            logger.info("Camera capture started")
        else:
            logger.error("Failed to start camera capture")
        
        # Initialize microphone
        mic = MicCapture()
        if mic.start():
            logger.info("Microphone capture started")
        else:
            logger.error("Failed to start microphone capture")
        
        # Initialize prompt engine
        prompt_engine = GemmaPromptEngine()
        if prompt_engine.start():
            logger.info("Gemma prompt engine started")
        else:
            logger.error("Failed to start Gemma prompt engine")
        
        # Initialize music generator
        music_generator = MagentaGeneration()
        if music_generator.start():
            logger.info("Magenta RT music generator started")
        else:
            logger.error("Failed to start Magenta RT music generator")
        
        # Initialize clip scheduler
        clip_scheduler = ClipScheduler()
        if clip_scheduler.start():
            logger.info("Clip scheduler started")
        else:
            logger.error("Failed to start clip scheduler")
        
        logger.info("All runtime components initialized")
        
    except Exception as e:
        logger.error(f"Error initializing runtime components: {e}")

@app.on_event("shutdown")
async def shutdown_event():
    """Cleanup runtime components on shutdown."""
    global camera, mic, prompt_engine, music_generator, clip_scheduler
    logger.info("Shutting down Synesthesia runtime components...")
    
    if camera:
        camera.stop()
    if mic:
        mic.stop()
    if prompt_engine:
        prompt_engine.stop()
    if music_generator:
        music_generator.stop()
    if clip_scheduler:
        clip_scheduler.stop()
    
    logger.info("All runtime components stopped")

@app.get("/")
async def root():
    """Root endpoint."""
    return {"message": "Synesthesia Runtime API", "version": "0.1.0"}

@app.get("/status")
async def get_status():
    """Get status of all runtime components."""
    status = {
        "camera": camera.is_running() if camera else False,
        "mic": mic.is_running() if mic else False,
        "prompt_engine": prompt_engine.is_running() if prompt_engine else False,
        "music_generator": music_generator.is_running() if music_generator else False,
        "clip_scheduler": clip_scheduler.is_running() if clip_scheduler else False,
    }
    
    if clip_scheduler:
        status["scheduler_details"] = clip_scheduler.get_status()
    
    return status

# WebSocket endpoint for real-time metrics and updates
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            # Send periodic updates
            await asyncio.sleep(1)
            # In a real implementation, we would send actual metrics here
            await manager.send_personal_message(json.dumps({"type": "ping"}), websocket)
    except WebSocketDisconnect:
        manager.disconnect(websocket)

# Model management endpoints
@app.post("/models/load")
async def load_model(model_name: str, precision: str = "4bit"):
    """Load a model (placeholder)."""
    # In a real implementation, this would trigger model loading
    return {"message": f"Model {model_name} with precision {precision} loading initiated"}

@app.post("/models/unload")
async def unload_model(model_name: str):
    """Unload a model (placeholder)."""
    return {"message": f"Model {model_name} unloading initiated"}

@app.get("/models/list")
async def list_models():
    """List available models (placeholder)."""
    return {
        "models": [
            {"name": "gemma-3n-e2b", "precisions": ["4bit", "8bit", "fp16"]},
            {"name": "gemma-3n-e4b", "precisions": ["4bit", "8bit", "fp16"]},
            {"name": "magenta-rt-small", "precisions": ["4bit", "8bit", "fp16"]},
            {"name": "magenta-rt-large", "precisions": ["4bit", "8bit", "fp16"]}
        ]
    }

# Generation control endpoints
@app.post("/generation/start")
async def start_generation():
    """Start continuous music generation."""
    # In a real implementation, this would trigger the generation loop
    return {"message": "Continuous generation started"}

@app.post("/generation/stop")
async def stop_generation():
    """Stop continuous music generation."""
    # In a real implementation, this would stop the generation loop
    return {"message": "Continuous generation stopped"}

@app.post("/generation/prompt")
async def inject_prompt(prompt: str):
    """Inject a custom prompt for the next generation."""
    if clip_scheduler:
        clip_scheduler.submit_prompt(prompt)
        return {"message": f"Prompt injected: {prompt[:50]}..."}
    else:
        raise HTTPException(status_code=503, detail="Clip scheduler not available")

# Metrics endpoints
@app.get("/metrics/current")
async def get_current_metrics():
    """Get current performance metrics."""
    metrics = {}
    if music_generator:
        metrics["avg_generation_time"] = music_generator.get_average_generation_time()
    if clip_scheduler:
        metrics.update(clip_scheduler.get_status())
    return metrics

if __name__ == "__main__":
    uvicorn.run("api.server:app", host="0.0.0.0", port=8000, reload=True)