zhanting_AI_MCP/calculator.py

# server.py
from mcp.server.fastmcp import FastMCP
import sys
import logging
import os
from typing import List, Dict, Any, Optional
import requests
from utils.logger_config import logger
try:
    import yaml
except ImportError:  # 允许在未安装 PyYAML 的环境下给出清晰日志
    yaml = None

# Fix UTF-8 encoding for Windows console
if sys.platform == 'win32':
    sys.stderr.reconfigure(encoding='utf-8')
    sys.stdout.reconfigure(encoding='utf-8')

# Create an MCP server
mcp = FastMCP("展厅交互模块")

# YAML 配置路径
VIDEO_CONFIG_PATH = os.path.join(os.path.dirname(__file__), 'video_config_prod.yaml')
CONFIG_PATH = os.path.join(os.path.dirname(__file__), 'config.yaml')

# 内存状态
_is_running: bool = False
_current_video_id: Optional[int] = None

def _load_config(config_path: str) -> Dict[str, Any]:
    if yaml is None:
        return {}
    if not os.path.exists(config_path):
        return {}
    try:
        with open(config_path, 'r', encoding='utf-8') as f:
            return yaml.safe_load(f) or {}
    except Exception as e:
        logger.error(f"读取配置文件失败 {config_path}: {e}")
        return {}

def _load_videos_from_yaml(config_path: str) -> List[Dict[str, Any]]:
    if yaml is None:
        logger.error("未安装 PyYAML，请在环境中安装 pyyaml 以读取配置文件")
        return []
    if not os.path.exists(config_path):
        logger.error(f"未找到配置文件: {config_path}")
        return []
    try:
        with open(config_path, 'r', encoding='utf-8') as f:
            data = yaml.safe_load(f) or {}
        videos = data.get('video_infos', [])
        if not isinstance(videos, list):
            logger.error("配置文件格式错误: 'video_infos' 应为列表")
            return []
        # 规范化字段
        normalized = []
        for v in videos:
            if not isinstance(v, dict):
                continue
            if 'id' not in v or 'name' not in v:
                continue
            normalized.append({
                'id': v.get('id'),
                'name': v.get('name'),
                'formula': v.get('formula'),
                'description': v.get('description'),
                'video_duration': v.get('video_duration'),
                'video_path': v.get('video_path'),
                'code': v.get('code'),
            })
        return normalized
    except Exception as e:
        logger.exception(f"读取配置失败: {e}")
        return []

def _index_videos(videos: List[Dict[str, Any]]) -> Dict[int, Dict[str, Any]]:
    return {int(v['id']): v for v in videos if isinstance(v.get('id'), int)}

# 启动时加载
_videos: List[Dict[str, Any]] = _load_videos_from_yaml(VIDEO_CONFIG_PATH)
_videos_by_id: Dict[int, Dict[str, Any]] = _index_videos(_videos)

# Flask API 基础地址（与 README_API.md 一致）
_app_config = _load_config(CONFIG_PATH)
FLASK_BASE = _app_config.get('flask_api_base', os.environ.get('FLASK_API_BASE', 'http://192.168.254.242:5050'))

def _flask_get(path: str) -> Dict[str, Any]:
    url = f"{FLASK_BASE}{path}"
    try:
        resp = requests.get(url, timeout=3)
        resp.raise_for_status()
        return resp.json()
    except Exception as e:
        logger.error(f"GET {url} 失败: {e}")
        return {"success": False, "message": str(e)}

def _flask_post(path: str, json_body: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    url = f"{FLASK_BASE}{path}"
    try:
        resp = requests.post(url, json=json_body or {}, timeout=3)
        resp.raise_for_status()
        return resp.json()
    except Exception as e:
        logger.error(f"POST {url} 失败: {e}")
        return {"success": False, "message": str(e)}

@mcp.tool(name="获取所有可用的视频列表")
def list_videos() -> dict:
    """获取所有可用的视频列表（来自 video_config.yaml）。

    使用建议：
    - 当用户只说出展品名称且可能存在谐音/近音时，先调用本工具获取候选列表，
      在客户端侧做相似度/拼音/编辑距离匹配，选取"最接近"的名称对应的 video_id，
      再调用 kodi_start(video_id)。
    
    返回的每个视频包含以下字段，可用于匹配：
    - id: 视频ID（整数）
    - name: 展品名称（如"氯化铷"、"硫酸钙新材料"等）
    - code: 展品编号（如"1号展品"、"2号展品"等）
    - formula: 化学式或公式
    - description: 描述信息
    """
    logger.info(f"获取视频列表，共 {_videos.__len__()} 个")
    return {"success": True, "videos": _videos, "count": len(_videos)}

@mcp.tool(name="播放到指定的展品视频")
def kodi_start(video_id: int) -> dict:
    """启动或切换 Kodi 播放到指定 video_id（转发到 Flask: POST /api/kodi/start）。

    参数：
    - video_id: 整数类型的视频ID（从 list_videos() 获取）

    使用场景和匹配策略：
    
    1. 用户说编号类表达（如"一号展品"、"1号产品"、"展品1"、"产品1"等）：
       - 提取数字部分（如"一号"→1、"1号"→1、"展品1"→1）
       - 直接使用该数字作为 video_id
    
    2. 用户直接说物质名称（如"播放氯化铷"、"播放硫酸钙"、"氯化铯介绍视频"等）：
       - 先调用 list_videos() 获取所有视频列表
       - 从返回的列表中匹配：
         a) 优先匹配 name 字段（精确或模糊匹配）
         b) 其次匹配 formula 字段
         c) 再次匹配 code 字段中的名称部分
       - 匹配时考虑：
         * 精确匹配（如"氯化铷"完全匹配）
         * 包含匹配（如输入"氯化铷"匹配到 name="氯化铷"）
         * 拼音/谐音匹配（如"氯化铯"与"氯化铷"的相似度）
         * 编辑距离（允许轻微口误）
       - 选择"最接近"的条目，使用其 id 字段作为 video_id 调用本接口
    
    3. 匹配示例：
       - "播放1号展品" → video_id=1
       - "播放一号产品" → video_id=1
       - "播放氯化铷" → 匹配 name="氯化铷" → video_id=1
       - "播放氯化铯介绍视频" → 匹配 name="氯化铯" → video_id=6
       - "硫酸钙" → 匹配 name="硫酸钙新材料"（包含匹配） → video_id=2
    
    提示：如果用户给的是展品"名称"而非 id，且存在谐音/口误，
    - 先调用 list_videos() 获取所有视频元数据
    - 在调用侧进行名称相似度匹配（可用拼音/编辑距离/分词等方法）
    - 选择"最接近"的条目并传入其 video_id 调用本接口
    """
    if not isinstance(video_id, int) or video_id < 0:
        return {"success": False, "message": "video_id 必须为大于等于 0 的整数"}
    if video_id not in _videos_by_id:
        return {"success": False, "message": f"未找到视频ID={video_id}"}
    video = _videos_by_id[video_id]
    logger.info(f"请求 Flask 启动/切换播放: 视频ID={video_id}，名称={video.get('name')}")
    api = _flask_post('/api/kodi/start', {"video_id": video_id})
    # 同步内存状态
    if api.get('success') is True:
        global _is_running, _current_video_id
        _is_running = True
        _current_video_id = video_id
    return api

@mcp.tool(name="设置电视播放视频音量全局音量")
def set_global_volume(volume: int) -> dict:
    """设置电视播放视频的全局音量 (转发到 Flask: POST /api/kodi/set_volume)。

    参数:
    - volume: 音量值，整数 0-100
    """
    # 参数校验
    if not isinstance(volume, int):
        try:
            volume = int(volume)
        except (ValueError, TypeError):
            return {"success": False, "message": "音量值必须为整数"}
            
    if volume < 0 or volume > 100:
        return {"success": False, "message": "音量值必须在 0-100 之间"}

    logger.info(f"请求设置全局音量: {volume}")
    return _flask_post('/api/kodi/set_volume', {"volume": volume})

@mcp.tool(name="打开办公楼大门")
def open_door(door_id: int, password: str = "") -> dict:
    """打开办公楼大门 (转发到 Flask: POST /api/door/open)。

    参数:
    - door_id: 门ID，整数 (必填)
    - password: 密码 (可选，默认为空字符串)
    """
    if not isinstance(door_id, int):
         return {"success": False, "message": "door_id 必须为整数"}

    logger.info(f"请求打开大门: door_id={door_id}")
    return _flask_post('/api/door/open', {"door_id": door_id, "password": password})

@mcp.tool(name="设置办公楼大门模式")
def set_door_mode(control_way: int, password: str = "") -> dict:
    """设置办公楼大门模式 (转发到 Flask: POST /api/door/control)。

    参数:
    - control_way: 模式，整数 (0:正常模式, 1:常开模式, 2:常闭模式)
    - password: 密码 (可选，默认为空字符串)
    """
    if not isinstance(control_way, int):
        return {"success": False, "message": "control_way 必须为整数"}
    
    if control_way not in [0, 1, 2]:
        return {"success": False, "message": "control_way 必须为 0, 1 或 2"}

    logger.info(f"请求设置大门模式: control_way={control_way}")
    return _flask_post('/api/door/control', {"control_way": control_way, "password": password})

@mcp.tool(name="打开指定电视")
def turn_on_tv(kodi_id: int) -> dict:
    """打开指定 ID 的电视 (转发到 Flask: POST /api/mitv/turn_on)。

    参数:
    - kodi_id: 整数 (必填) 对应 Kodi 客户端的 ID/索引

    使用说明：
    - 展厅共有 6 台电视，顺序从左到右。
    - kodi_id 从 0 开始计数：
      * "第一台"、"1号电视"、"左边第一台" -> kodi_id=0
      * "第二台"、"2号电视" -> kodi_id=1
      * ...
      * "第六台"、"6号电视"、"最右边那台" -> kodi_id=5
    """
    if not isinstance(kodi_id, int):
        return {"success": False, "message": "kodi_id 必须为整数"}

    logger.info(f"请求打开电视: kodi_id={kodi_id}")
    return _flask_post('/api/mitv/turn_on', {"kodi_id": kodi_id})

@mcp.tool(name="关闭指定电视")
def turn_off_tv(kodi_id: int) -> dict:
    """关闭指定 ID 的电视 (转发到 Flask: POST /api/mitv/turn_off)。

    参数:
    - kodi_id: 整数 (必填) 对应 Kodi 客户端的 ID/索引

    使用说明：
    - 展厅共有 6 台电视，顺序从左到右。
    - kodi_id 从 0 开始计数：
      * "第一台"、"1号电视"、"左边第一台" -> kodi_id=0
      * "第二台"、"2号电视" -> kodi_id=1
      * ...
      * "第六台"、"6号电视"、"最右边那台" -> kodi_id=5
    """
    if not isinstance(kodi_id, int):
        return {"success": False, "message": "kodi_id 必须为整数"}

    logger.info(f"请求关闭电视: kodi_id={kodi_id}")
    return _flask_post('/api/mitv/turn_off', {"kodi_id": kodi_id})

@mcp.tool(name="打开所有电视")
def turn_on_all_tvs() -> dict:
    """打开所有电视 (转发到 Flask: POST /api/mitv/turn_on_all)。"""
    logger.info("请求打开所有电视")
    return _flask_post('/api/mitv/turn_on_all')

@mcp.tool(name="关闭所有电视")
def turn_off_all_tvs() -> dict:
    """关闭所有电视 (转发到 Flask: POST /api/mitv/turn_off_all)。"""
    logger.info("请求关闭所有电视")
    return _flask_post('/api/mitv/turn_off_all')

# 灯光控制相关工具
@mcp.tool(name="控制一楼大门玄关顶灯")
def control_entrance_lights(action: str) -> dict:
    """控制一楼大门玄关顶灯 (转发到 Flask: POST /api/ha/entrance_lights/turn_on|off)。

    参数:
    - action: "turn_on" (打开) 或 "turn_off" (关闭)
    """
    if action not in ["turn_on", "turn_off"]:
        return {"success": False, "message": "action 必须为 'turn_on' 或 'turn_off'"}
    
    logger.info(f"请求控制玄关顶灯: {action}")
    return _flask_post(f'/api/ha/entrance_lights/{action}')

@mcp.tool(name="控制一楼大门玄关射灯")
def control_exhibition_spotlight(action: str) -> dict:
    """控制一楼大门玄关射灯 (转发到 Flask: POST /api/ha/exhibition_spotlight/turn_on|off)。

    参数:
    - action: "turn_on" (打开) 或 "turn_off" (关闭)
    """
    if action not in ["turn_on", "turn_off"]:
        return {"success": False, "message": "action 必须为 'turn_on' 或 'turn_off'"}
    
    logger.info(f"请求控制玄关射灯: {action}")
    return _flask_post(f'/api/ha/exhibition_spotlight/{action}')

@mcp.tool(name="控制一楼展厅顶灯")
def control_exhibition_ceiling_light(action: str) -> dict:
    """控制一楼展厅顶灯 (转发到 Flask: POST /api/ha/exhibition_ceiling_light/turn_on|off)。

    参数:
    - action: "turn_on" (打开) 或 "turn_off" (关闭)
    """
    if action not in ["turn_on", "turn_off"]:
        return {"success": False, "message": "action 必须为 'turn_on' 或 'turn_off'"}
    
    logger.info(f"请求控制展厅顶灯: {action}")
    return _flask_post(f'/api/ha/exhibition_ceiling_light/{action}')

@mcp.tool(name="控制展厅桌面灯座总开关")
def control_exhibition_desktop_switch(action: str) -> dict:
    """控制展厅桌面灯座总开关 (转发到 Flask: POST /api/ha/exhibition_desktop_switch/turn_on|off)。

    参数:
    - action: "turn_on" (打开) 或 "turn_off" (关闭)
    """
    if action not in ["turn_on", "turn_off"]:
        return {"success": False, "message": "action 必须为 'turn_on' 或 'turn_off'"}
    
    logger.info(f"请求控制桌面灯座总开关: {action}")
    return _flask_post(f'/api/ha/exhibition_desktop_switch/{action}')

@mcp.tool(name="控制展厅桌面3D风扇投影")
def control_exhibition_3d_fan(action: str) -> dict:
    """控制展厅桌面3D风扇投影 (转发到 Flask: POST /api/ha/exhibition_3d_fan/turn_on|off)。

    参数:
    - action: "turn_on" (打开) 或 "turn_off" (关闭)
    """
    if action not in ["turn_on", "turn_off"]:
        return {"success": False, "message": "action 必须为 'turn_on' 或 'turn_off'"}
    
    logger.info(f"请求控制3D风扇投影: {action}")
    return _flask_post(f'/api/ha/exhibition_3d_fan/{action}')

@mcp.tool(name="控制展台桌子灯带")
def control_exhibition_stand_light_strip(action: str) -> dict:
    """控制展台桌子灯带 (转发到 Flask: POST /api/ha/exhibition_stand_light_strip/turn_on|off)。

    参数:
    - action: "turn_on" (打开) 或 "turn_off" (关闭)
    """
    if action not in ["turn_on", "turn_off"]:
        return {"success": False, "message": "action 必须为 'turn_on' 或 'turn_off'"}
    
    logger.info(f"请求控制展台灯带: {action}")
    return _flask_post(f'/api/ha/exhibition_stand_light_strip/{action}')

# Start the server
if __name__ == "__main__":
    mcp.run(transport="stdio")