import base64 import hashlib import json import queue import socket import threading from http import HTTPStatus from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Callable from ..bridge import WakewordEventBridge from ..config.config_loader import load_config class TestRuntimeHttpServer: def __init__(self, test_root: Path, host: str = "0.0.0.0", port: int = 8006) -> None: self.test_root = test_root self.host = host self.port = port self.event_bridge = WakewordEventBridge() self._restart_handler: Callable[[], None] | None = None self._restart_lock = threading.Lock() self._server = self._build_server() @property def page_url(self) -> str: return f"http://127.0.0.1:{self.port}/index.html" @property def bridge_url(self) -> str: return f"ws://127.0.0.1:{self.port}/wakeword-ws" def serve_forever(self) -> None: self._server.serve_forever() def shutdown(self) -> None: self._server.shutdown() self.event_bridge.close() self._server.server_close() def set_restart_handler(self, handler: Callable[[], None]) -> None: self._restart_handler = handler def request_runtime_restart(self) -> None: with self._restart_lock: handler = self._restart_handler if handler is None: raise RuntimeError("restart handler is not configured") threading.Thread( target=self._run_restart_handler, name="test-runtime-restart", daemon=True, ).start() def _run_restart_handler(self) -> None: handler = self._restart_handler if handler is None: return handler() def _build_server(self) -> ThreadingHTTPServer: test_root = self.test_root event_bridge = self.event_bridge schedule_restart = self.request_runtime_restart class TestRuntimeHandler(SimpleHTTPRequestHandler): protocol_version = "HTTP/1.1" def __init__(self, *args, **kwargs): super().__init__(*args, directory=str(test_root), **kwargs) def handle(self) -> None: try: super().handle() except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError): pass def do_GET(self) -> None: if self.path == "/wakeword-ws": self._handle_websocket(event_bridge) return if self.path == "/health": body = json.dumps({"status": "ok"}).encode("utf-8") self.send_response(HTTPStatus.OK) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Cache-Control", "no-cache") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return super().do_GET() def log_message(self, format: str, *args) -> None: return def _handle_websocket(self, bridge: WakewordEventBridge) -> None: if self.headers.get("Upgrade", "").lower() != "websocket": self.send_error(HTTPStatus.BAD_REQUEST, "expected websocket upgrade") return websocket_key = self.headers.get("Sec-WebSocket-Key") if not websocket_key: self.send_error(HTTPStatus.BAD_REQUEST, "missing Sec-WebSocket-Key") return accept_source = websocket_key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" accept_value = base64.b64encode( hashlib.sha1(accept_source.encode("utf-8")).digest() ).decode("ascii") client_queue = bridge.add_client() self.send_response(HTTPStatus.SWITCHING_PROTOCOLS) self.send_header("Upgrade", "websocket") self.send_header("Connection", "Upgrade") self.send_header("Sec-WebSocket-Accept", accept_value) self.end_headers() try: self.connection.settimeout(0.2) self._send_websocket_text(bridge.build_ready_message()) self._send_websocket_text(self._build_wakeword_config_message(bridge)) while bridge.is_running: inbound_message = self._receive_websocket_message() if inbound_message is not None: response_message = self._handle_bridge_request(bridge, inbound_message) if response_message: self._send_websocket_text(response_message) try: message = client_queue.get(timeout=0.2) if message == "__bridge_closed__": break self._send_websocket_text(message) except queue.Empty: if not bridge.is_running: break continue except socket.timeout: pass except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError): pass finally: bridge.remove_client(client_queue) def _build_wakeword_config_message(self, bridge: WakewordEventBridge) -> str: try: runtime_root = test_root / "wakeword_runtime" config = load_config(runtime_root) payload = { "enabled": config.wakeword_enabled, "wakeWords": config.wake_words, } return bridge.build_message("wakeword_config", payload) except Exception as exc: return bridge.build_message( "wakeword_config", {}, success=False, error=f"读取唤醒词配置失败: {exc}", ) def _handle_bridge_request(self, bridge: WakewordEventBridge, raw_message: str) -> str | None: try: message = json.loads(raw_message) except json.JSONDecodeError: return None message_type = str(message.get("type", "")).strip() request_id = message.get("requestId") payload = message.get("payload") or {} result_type = f"{message_type}_result" if message_type else "bridge_request_result" if message_type == "set_wakeword_config": try: result_payload = self._save_wakeword_config(payload) bridge.publish("wakeword_config", result_payload) return bridge.build_message( "set_wakeword_config_result", result_payload, request_id=request_id, ) except Exception as exc: return bridge.build_message( "set_wakeword_config_result", {}, request_id=request_id, success=False, error=f"保存唤醒词配置失败: {exc}", ) if message_type == "restart_wakeword_service": schedule_restart() return bridge.build_message( "restart_wakeword_service_result", {"restarting": True}, request_id=request_id, ) return bridge.build_message( result_type, {}, request_id=request_id, success=False, error=f"unsupported message type: {message_type}", ) def _save_wakeword_config(self, payload: dict) -> dict: runtime_root = test_root / "wakeword_runtime" config_path = runtime_root / "config.json" model_root = runtime_root / "models" keywords_path = model_root / "keywords.txt" enabled = bool(payload.get("enabled", True)) wake_words = payload.get("wakeWords") or [] normalized_wake_words = [] for item in wake_words: if not isinstance(item, str): continue text = item.strip() if text and text not in normalized_wake_words: normalized_wake_words.append(text) if enabled and not normalized_wake_words: raise ValueError("wakeWords cannot be empty when wakeword is enabled") raw_config = json.loads(config_path.read_text(encoding="utf-8")) raw_config.setdefault("wakeword", {})["enabled"] = enabled config_path.write_text( json.dumps(raw_config, indent=2, ensure_ascii=False), encoding="utf-8", ) keywords_lines = [self._build_keyword_line(item) for item in normalized_wake_words] keywords_path.write_text( ("\n".join(keywords_lines) + "\n") if keywords_lines else "", encoding="utf-8", ) return { "enabled": enabled, "wakeWords": normalized_wake_words, } def _build_keyword_line(self, keyword_text: str) -> str: from pypinyin import Style, pinyin initials = pinyin(keyword_text, style=Style.INITIALS, strict=False) finals = pinyin( keyword_text, style=Style.FINALS_TONE, strict=False, neutral_tone_with_five=True, ) tokens: list[str] = [] for initial_parts, final_parts in zip(initials, finals): initial = initial_parts[0].strip() final = final_parts[0].strip() if initial: tokens.append(initial) if final: tokens.append(final) if not tokens: raise ValueError(f"failed to generate pinyin tokens for wake word: {keyword_text}") return f"{' '.join(tokens)} @{keyword_text}" def _receive_websocket_message(self) -> str | None: try: header = self._read_exact(2) except socket.timeout: return None if not header: return None first_byte, second_byte = header[0], header[1] opcode = first_byte & 0x0F masked = (second_byte & 0x80) != 0 payload_length = second_byte & 0x7F if payload_length == 126: payload_length = int.from_bytes(self._read_exact(2), "big") elif payload_length == 127: payload_length = int.from_bytes(self._read_exact(8), "big") masking_key = self._read_exact(4) if masked else b"" payload = self._read_exact(payload_length) if payload_length else b"" if masked and payload: payload = bytes( byte ^ masking_key[index % 4] for index, byte in enumerate(payload) ) if opcode == 0x8: raise ConnectionAbortedError("websocket closed by client") if opcode == 0x9: self._send_websocket_frame(0xA, payload) return None if opcode == 0xA: return None if opcode != 0x1: return None return payload.decode("utf-8") def _read_exact(self, size: int) -> bytes: if size <= 0: return b"" chunks = bytearray() while len(chunks) < size: chunk = self.connection.recv(size - len(chunks)) if not chunk: raise ConnectionResetError("websocket connection closed") chunks.extend(chunk) return bytes(chunks) def _send_websocket_text(self, message: str) -> None: self._send_websocket_frame(0x1, message.encode("utf-8")) def _send_websocket_frame(self, opcode: int, payload: bytes) -> None: header = bytearray() header.append(0x80 | opcode) payload_length = len(payload) if payload_length < 126: header.append(payload_length) elif payload_length < 65536: header.append(126) header.extend(payload_length.to_bytes(2, "big")) else: header.append(127) header.extend(payload_length.to_bytes(8, "big")) self.wfile.write(bytes(header) + payload) self.wfile.flush() server = ThreadingHTTPServer((self.host, self.port), TestRuntimeHandler) server.daemon_threads = True return server