mcp/camera_server.py

"""
MCP Camera Tool Server — JSON-RPC style interface for camera control.

Exposes camera management, VQA, alert rules, and alert history as
callable tools that can be consumed by any MCP-compatible client.

Tools:
- list_cameras: Enumerate all registered cameras and their status
- get_camera_status: Detailed health info for a single camera
- get_latest_frame: Retrieve the most recent frame as base64 JPEG
- ask_camera: Visual question answering on the latest frame
- set_alert_rule: Register an alert rule on a camera
- get_alert_history: Retrieve recent alert events for a camera
"""

from __future__ import annotations

import base64
import logging
import time
from typing import Any, Callable

logger = logging.getLogger(__name__)


class CameraToolServer:
    """Camera control tools exposed via MCP protocol.

    Implements a JSON-RPC style request handler that routes method calls
    to individual tool functions. Designed to work with any camera manager
    (edge.ingest.CameraManager) and model manager for VQA inference.

    Args:
        camera_manager: CameraManager instance (from edge/ingest.py).
        model_manager: Object with an ``ask(image, question)`` method for VQA.
    """

    def __init__(self, camera_manager=None, model_manager=None):
        self.cameras = camera_manager
        self.model = model_manager

        # Alert rules stored per camera: {camera_id: [rule_dicts]}
        self._alert_rules: dict[str, list[dict[str, Any]]] = {}

        # Alert event history: list of alert dicts (most recent last)
        self._alert_history: list[dict[str, Any]] = []

        # Tool registry
        self.tools: dict[str, Callable] = {
            "list_cameras": self.list_cameras,
            "get_camera_status": self.get_camera_status,
            "get_latest_frame": self.get_latest_frame,
            "ask_camera": self.ask_camera,
            "set_alert_rule": self.set_alert_rule,
            "get_alert_history": self.get_alert_history,
        }

    # ------------------------------------------------------------------
    # Tool descriptors (for MCP tool listing)
    # ------------------------------------------------------------------

    def get_tool_definitions(self) -> list[dict[str, Any]]:
        """Return MCP-compatible tool definitions."""
        return [
            {
                "name": "list_cameras",
                "description": "List all registered cameras and their current status.",
                "parameters": {},
            },
            {
                "name": "get_camera_status",
                "description": "Get detailed status for a specific camera.",
                "parameters": {
                    "camera_id": {"type": "string", "description": "Camera identifier."},
                },
            },
            {
                "name": "get_latest_frame",
                "description": "Get the latest frame from a camera as base64 JPEG.",
                "parameters": {
                    "camera_id": {"type": "string", "description": "Camera identifier."},
                },
            },
            {
                "name": "ask_camera",
                "description": "Ask a visual question about the latest frame from a camera.",
                "parameters": {
                    "camera_id": {"type": "string", "description": "Camera identifier."},
                    "question": {"type": "string", "description": "Natural language question."},
                },
            },
            {
                "name": "set_alert_rule",
                "description": "Register an alert rule on a camera.",
                "parameters": {
                    "camera_id": {"type": "string", "description": "Camera identifier."},
                    "rule": {"type": "object", "description": "Alert rule definition."},
                },
            },
            {
                "name": "get_alert_history",
                "description": "Get recent alert events for a camera.",
                "parameters": {
                    "camera_id": {"type": "string", "description": "Camera identifier."},
                    "limit": {"type": "integer", "description": "Max events to return.", "default": 20},
                },
            },
        ]

    # ------------------------------------------------------------------
    # Tools
    # ------------------------------------------------------------------

    def list_cameras(self, **kwargs) -> dict[str, Any]:
        """List all registered cameras with status summary."""
        if self.cameras is None:
            return {"cameras": [], "count": 0}

        camera_ids = self.cameras.camera_ids()
        status_map = self.cameras.status()

        cameras = []
        for cid in camera_ids:
            health = status_map.get(cid, {})
            cameras.append({
                "camera_id": cid,
                "state": health.get("state", "unknown"),
                "fps": health.get("fps_actual", 0.0),
                "frames_captured": health.get("frames_captured", 0),
            })

        return {"cameras": cameras, "count": len(cameras)}

    def get_camera_status(self, camera_id: str, **kwargs) -> dict[str, Any]:
        """Get detailed health info for a single camera."""
        if self.cameras is None:
            return {"error": "No camera manager configured"}

        status_map = self.cameras.status()
        health = status_map.get(camera_id)
        if health is None:
            return {"error": f"Camera '{camera_id}' not found"}

        return {
            "camera_id": camera_id,
            **health,
        }

    def get_latest_frame(self, camera_id: str, **kwargs) -> dict[str, Any]:
        """Return the latest frame from a camera as base64-encoded JPEG."""
        if self.cameras is None:
            return {"error": "No camera manager configured"}

        result = self.cameras.get_frame(camera_id)
        if result is None:
            return {"error": f"No frame available for camera '{camera_id}'"}

        frame, timestamp = result

        # Encode frame to JPEG then base64
        try:
            import cv2
            _, jpeg_buf = cv2.imencode(".jpg", frame)
            b64 = base64.b64encode(jpeg_buf.tobytes()).decode("utf-8")
        except ImportError:
            # Fallback: raw numpy bytes (unlikely in production)
            b64 = base64.b64encode(frame.tobytes()).decode("utf-8")

        return {
            "camera_id": camera_id,
            "frame_base64": b64,
            "format": "jpeg",
            "timestamp": timestamp,
        }

    def ask_camera(self, camera_id: str, question: str, **kwargs) -> dict[str, Any]:
        """Run visual question answering on the latest frame from a camera."""
        if self.cameras is None:
            return {"error": "No camera manager configured"}
        if self.model is None:
            return {"error": "No model manager configured"}

        result = self.cameras.get_frame(camera_id)
        if result is None:
            return {"error": f"No frame available for camera '{camera_id}'"}

        frame, timestamp = result

        try:
            answer = self.model.ask(frame, question)
        except Exception as exc:
            logger.error("ask_camera failed for %s: %s", camera_id, exc)
            return {"error": f"Model inference failed: {exc}"}

        return {
            "camera_id": camera_id,
            "question": question,
            "answer": answer,
            "timestamp": timestamp,
        }

    def set_alert_rule(self, camera_id: str, rule: dict, **kwargs) -> dict[str, Any]:
        """Register an alert rule for a camera."""
        rule_id = rule.get("rule_id", f"rule-{int(time.time() * 1000)}")
        rule["rule_id"] = rule_id

        if camera_id not in self._alert_rules:
            self._alert_rules[camera_id] = []

        self._alert_rules[camera_id].append(rule)

        logger.info("Set alert rule '%s' on camera '%s'", rule_id, camera_id)
        return {
            "camera_id": camera_id,
            "rule_id": rule_id,
            "status": "created",
            "total_rules": len(self._alert_rules[camera_id]),
        }

    def get_alert_history(self, camera_id: str, limit: int = 20, **kwargs) -> dict[str, Any]:
        """Get recent alert events for a camera."""
        events = [
            e for e in self._alert_history
            if e.get("camera_id") == camera_id
        ]
        # Most recent first, limited
        events = list(reversed(events))[:limit]

        return {
            "camera_id": camera_id,
            "events": events,
            "count": len(events),
        }

    def record_alert(self, event: dict[str, Any]) -> None:
        """Record an alert event to history (called by alert processing pipeline)."""
        if "timestamp" not in event:
            event["timestamp"] = time.time()
        self._alert_history.append(event)

    # ------------------------------------------------------------------
    # JSON-RPC style request handler
    # ------------------------------------------------------------------

    def handle_request(self, method: str, params: dict | None = None) -> dict[str, Any]:
        """
        JSON-RPC style request handler.

        Args:
            method: Tool name to invoke.
            params: Keyword arguments for the tool.

        Returns:
            Dict with either ``result`` key on success or ``error`` key on failure.
        """
        if params is None:
            params = {}

        tool_fn = self.tools.get(method)
        if tool_fn is None:
            return {
                "error": {
                    "code": -32601,
                    "message": f"Method not found: {method}",
                },
            }

        try:
            result = tool_fn(**params)
            return {"result": result}
        except Exception as exc:
            logger.exception("Error executing tool '%s'", method)
            return {
                "error": {
                    "code": -32000,
                    "message": str(exc),
                },
            }