|
|
@@ -21,11 +21,24 @@ 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 以读取配置文件")
|
|
|
@@ -69,7 +82,8 @@ _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 一致)
|
|
|
-FLASK_BASE = os.environ.get('FLASK_API_BASE', 'http://192.168.254.242:5050')
|
|
|
+_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}"
|
|
|
@@ -162,6 +176,190 @@ def kodi_start(video_id: int) -> dict:
|
|
|
_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")
|