reyna-cli/backup/macmini-automation-baseline-2026-08-03/hermes-scripts/reynabot_screen_mcp.py

#!/usr/bin/env python3
"""Small MCP facade for the ReynaBot ESP32/Tactility screen.

The board exposes a lightweight JSON-RPC endpoint at /api/mcp but does not
complete Hermes native HTTP MCP initialization. This stdio MCP wrapper gives the
kids profile stable, child-safe screen/audio tools while proxying to the board.
"""
from __future__ import annotations

import base64
import io
import json
import math
import os
import struct
import sys
import urllib.error
import urllib.request
import wave
from pathlib import Path
from typing import Any

from mcp.server.fastmcp import FastMCP

SCREEN_URL = os.environ.get("REYNABOT_SCREEN_MCP_URL", "http://192.168.68.130/api/mcp")
HERMES_REPO = os.environ.get("HERMES_REPO", "/home/adolforeyna/.hermes/hermes-agent")
if HERMES_REPO not in sys.path:
    sys.path.insert(0, HERMES_REPO)

mcp = FastMCP("reynabot_screen")


def _call_board(tool_name: str, arguments: dict[str, Any], timeout: int = 30) -> str:
    payload = {
        "jsonrpc": "2.0",
        "id": "hermes",
        "method": "tools/call",
        "params": {"name": tool_name, "arguments": arguments},
    }
    data = json.dumps(payload).encode("utf-8")
    req = urllib.request.Request(SCREEN_URL, data=data, headers={"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            body = resp.read().decode("utf-8", "ignore")
    except urllib.error.HTTPError as exc:
        body = exc.read().decode("utf-8", "ignore")
        return f"HTTP {exc.code} from board: {body[:500]}"
    except Exception as exc:
        return f"Board call failed: {type(exc).__name__}: {exc}"
    try:
        parsed = json.loads(body)
    except Exception:
        return body[:1000]
    if parsed.get("error"):
        return f"Board error: {parsed['error']}"
    content = ((parsed.get("result") or {}).get("content") or [])
    texts = [str(item.get("text")) for item in content if isinstance(item, dict) and item.get("text")]
    return "\n".join(texts) if texts else json.dumps(parsed.get("result"), ensure_ascii=False)[:1000]


@mcp.tool()
def get_screen_capabilities() -> str:
    """Get the ReynaBot screen display/audio capabilities."""
    return _call_board("get_capabilities", {}, timeout=8)


@mcp.tool()
def draw_screen_text(text: str, x: int = 8, y: int = 8, size: int = 1, clear: bool = True) -> str:
    """Draw short text on the ReynaBot screen. Use this for simple visual replies."""
    if clear:
        _call_board("clear_screen", {"color": 0}, timeout=8)
    return _call_board("draw_text", {"text": text[:900], "x": x, "y": y, "size": 2 if size == 2 else 1}, timeout=8)


@mcp.tool()
def play_screen_tone(frequency: int = 440, duration_ms: int = 250, volume: int = 35) -> str:
    """Play a short tone on the ReynaBot speaker."""
    frequency = max(80, min(4000, int(frequency)))
    duration_ms = max(30, min(2000, int(duration_ms)))
    volume = max(0, min(100, int(volume)))
    return _call_board("play_tone", {"frequency": frequency, "duration_ms": duration_ms, "volume": volume}, timeout=8)


def _image_to_pbm_base64(ref: str) -> str:
    from PIL import Image, ImageOps

    ref = ref.strip()
    if ref.lower().startswith("data:image/"):
        _, _, encoded = ref.partition(",")
        raw = base64.b64decode(encoded)
    elif ref.lower().startswith(("http://", "https://")):
        with urllib.request.urlopen(ref, timeout=20) as resp:
            raw = resp.read()
    else:
        raw = Path(ref).expanduser().read_bytes()
    with Image.open(io.BytesIO(raw)) as img:
        img = ImageOps.exif_transpose(img).convert("RGB")
        canvas = Image.new("RGB", (320, 240), "white")
        img.thumbnail((320, 240), Image.Resampling.LANCZOS)
        canvas.paste(img, ((320 - img.width) // 2, (240 - img.height) // 2))
        mono = canvas.convert("1", dither=Image.Dither.FLOYDSTEINBERG)
        out = io.BytesIO()
        mono.save(out, format="PPM")  # mode=1 writes raw PBM (P4)
        return base64.b64encode(out.getvalue()).decode("ascii")


@mcp.tool()
def show_screen_image(image_url_or_path: str) -> str:
    """Display an image URL, data:image URL, or local image path on the ReynaBot screen.

    Use this after generating an image. The wrapper converts the image to a
    small 1-bit PBM so it fits the board's MCP request limits.
    """
    try:
        pbm_base64 = _image_to_pbm_base64(image_url_or_path)
    except Exception as exc:
        return f"Image conversion failed: {type(exc).__name__}: {exc}"
    return _call_board("draw_image", {"pbm_base64": pbm_base64, "x": 0, "y": 0, "dither": False}, timeout=15)


def _text_to_mp3_base64(text: str) -> str:
    from tools.tts_tool import text_to_speech_tool

    result = json.loads(text_to_speech_tool(text[:420]))
    if not result.get("success"):
        raise RuntimeError(result.get("error") or "TTS failed")
    path = Path(str(result.get("file_path") or ""))
    if not path.exists():
        raise RuntimeError("TTS file missing")
    if path.suffix.lower() != ".mp3":
        # Fall through to board WAV support only for short audio to avoid huge payloads.
        return ""
    return base64.b64encode(path.read_bytes()).decode("ascii")


def _short_wav_base64(text: str) -> str:
    # Last-resort tiny tone-like WAV if a TTS provider returns a non-MP3 file.
    sample_rate = 16000
    duration_s = min(0.8, max(0.15, len(text) / 80.0))
    buf = io.BytesIO()
    with wave.open(buf, "wb") as wav:
        wav.setnchannels(1)
        wav.setsampwidth(2)
        wav.setframerate(sample_rate)
        frames = bytearray()
        for i in range(int(sample_rate * duration_s)):
            sample = int(4000 * math.sin(2 * math.pi * 660 * i / sample_rate))
            frames += struct.pack("<h", sample)
        wav.writeframes(frames)
    return base64.b64encode(buf.getvalue()).decode("ascii")


@mcp.tool()
def speak_screen_text(text: str, volume: int = 70) -> str:
    """Speak short text through the ReynaBot speaker using the configured Hermes TTS voice."""
    volume = max(0, min(100, int(volume)))
    try:
        mp3_base64 = _text_to_mp3_base64(text)
        if mp3_base64:
            return _call_board("play_mp3_base64", {"mp3_base64": mp3_base64, "volume": volume}, timeout=75)
        return _call_board("play_audio_base64", {"wav_base64": _short_wav_base64(text), "volume": volume}, timeout=20)
    except Exception as exc:
        return f"Speech failed: {type(exc).__name__}: {exc}"


if __name__ == "__main__":
    mcp.run()