添加红外扩展

src/cognitive_agent/server.py

@@ -315,6 +315,15 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
         if path == "/api/extension/nfc/status":
             self._json({"success": True, **self.nfc.snapshot()})
             return
+        if path == "/api/extension/gscu/slots":
+            try:
+                self._json(self.extension.list_infrared_slots())
+            except GatewayError as exc:
+                self._json(
+                    {"success": False, "error": "catalog_unavailable", "message": str(exc)},
+                    HTTPStatus.SERVICE_UNAVAILABLE,
+                )
+            return
         if path == "/api/extension/gscu/info":
             self._handle_extension_gateway("info", {})
             return
@@ -902,11 +911,44 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
                 detail=str(exc),
             )
 
+    def do_PUT(self) -> None:  # noqa: N802 - stdlib handler API
+        path = unquote(urlparse(self.path).path)
+        if not self._authorize(path):
+            return
+        match = re.fullmatch(r"/api/extension/gscu/slots/(?P<slot>\d+)", path)
+        if match is None:
+            self._json({"error": "not_found"}, HTTPStatus.NOT_FOUND)
+            return
+        try:
+            payload = self._read_json_body(max_bytes=4096)
+            self._json(
+                self.extension.save_infrared_slot(match.group("slot"), payload),
+                HTTPStatus.OK,
+            )
+        except (GatewayError, ValueError, json.JSONDecodeError) as exc:
+            self._json(
+                {"success": False, "error": "invalid_request", "message": str(exc)},
+                HTTPStatus.BAD_REQUEST,
+            )
+
     def do_DELETE(self) -> None:  # noqa: N802 - stdlib handler API
         parsed_url = urlparse(self.path)
         path = unquote(parsed_url.path)
         if not self._authorize(path):
             return
+        slot_match = re.fullmatch(r"/api/extension/gscu/slots/(?P<slot>\d+)", path)
+        if slot_match is not None:
+            try:
+                self._json(
+                    self.extension.delete_infrared_slot(slot_match.group("slot")),
+                    HTTPStatus.OK,
+                )
+            except GatewayError as exc:
+                self._json(
+                    {"success": False, "error": "invalid_request", "message": str(exc)},
+                    HTTPStatus.BAD_REQUEST,
+                )
+            return
         prefix = "/api/mode-profiles/"
         if not path.startswith(prefix):
             self._json({"error": "not_found"}, HTTPStatus.NOT_FOUND)
@@ -966,18 +1008,46 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
 
     def _handle_extension_post(self, path: str) -> None:
         action_map = {
+            "/api/extension/gscu/info": "info",
             "/api/extension/gscu/learn": "learn",
             "/api/extension/gscu/cancel": "cancel",
             "/api/extension/gscu/emit": "emit",
             "/api/extension/gscu/read": "read",
-            "/api/extension/gscu/erase": "erase",
+            "/api/extension/gscu/write": "write",
+            "/api/extension/gscu/restart": "restart",
+            "/api/extension/gscu/reset": "reset",
+            "/api/extension/gscu/wifi-lock": "wifi_lock",
+            "/api/extension/gscu/raw": "raw",
         }
         try:
             payload = self._read_json_body(max_bytes=256 * 1024)
+            if path in {
+                "/api/extension/gscu/erase",
+                "/api/extension/gscu/erase-all",
+            }:
+                result = self.extension.erase_all_infrared(dict(payload))
+                self.vehicle_display_sync.update_extension(
+                    action={"type": "erase_all"},
+                    result=result,
+                )
+                self._json(result)
+                return
             action = action_map.get(path)
             if action:
                 self._handle_extension_gateway(action, payload)
                 return
+            if path == "/api/extension/gscu/config":
+                result = self.extension.configure_tcp_gateway(dict(payload))
+                self.vehicle_display_sync.update_extension(
+                    action={
+                        "type": "gateway_config",
+                        "listen_host": result.get("listen_host"),
+                        "port": result.get("port"),
+                    },
+                    result=result,
+                )
+                self._json(result)
+                return
             if path == "/api/extension/display-sync":
                 catalog = payload.get("peripherals", payload.get("catalog", []))
                 if not isinstance(catalog, list) or len(catalog) > 248: