“wangquan” 57fdf656b9 add_xiaozhi
2025-07-18 13:12:09 +08:00

72 lines
2.4 KiB
Python
Executable File
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import asyncio
from aiohttp import web
from config.logger import setup_logging
from core.api.ota_handler import OTAHandler
from core.api.vision_handler import VisionHandler
TAG = __name__
class SimpleHttpServer:
def __init__(self, config: dict):
self.config = config
self.logger = setup_logging()
self.ota_handler = OTAHandler(config)
self.vision_handler = VisionHandler(config)
def _get_websocket_url(self, local_ip: str, port: int) -> str:
"""获取websocket地址
Args:
local_ip: 本地IP地址
port: 端口号
Returns:
str: websocket地址
"""
server_config = self.config["server"]
websocket_config = server_config.get("websocket")
if websocket_config and "" not in websocket_config:
return websocket_config
else:
return f"ws://{local_ip}:{port}/xiaozhi/v1/"
async def start(self):
server_config = self.config["server"]
host = server_config.get("ip", "0.0.0.0")
port = int(server_config.get("http_port", 8003))
if port:
app = web.Application()
read_config_from_api = server_config.get("read_config_from_api", False)
if not read_config_from_api:
# 如果没有开启智控台只是单模块运行就需要再添加简单OTA接口用于下发websocket接口
app.add_routes(
[
web.get("/xiaozhi/ota/", self.ota_handler.handle_get),
web.post("/xiaozhi/ota/", self.ota_handler.handle_post),
web.options("/xiaozhi/ota/", self.ota_handler.handle_post),
]
)
# 添加路由
app.add_routes(
[
web.get("/mcp/vision/explain", self.vision_handler.handle_get),
web.post("/mcp/vision/explain", self.vision_handler.handle_post),
web.options("/mcp/vision/explain", self.vision_handler.handle_post),
]
)
# 运行服务
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, host, port)
await site.start()
# 保持服务运行
while True:
await asyncio.sleep(3600) # 每隔 1 小时检查一次