Miayui
367 lines
18 KiB
Python
367 lines
18 KiB
Python
"""Deterministic bridge from CAN_DEV plans to the trusted native CAN API.
|
|
|
|
The mapping in this module is intentionally equivalent to CAN_DEV's
|
|
``apps/web-console/app.js::cxxCanCommandsForOperation``. Natural language is
|
|
never forwarded to the hardware gateway. Only structured capabilities are
|
|
converted to the gateway's bounded command grammar, validated as a complete
|
|
batch, and then (when explicitly enabled) executed in order.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urlparse
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
class NativeCanError(RuntimeError):
|
|
"""Raised when native CAN validation or execution cannot complete safely."""
|
|
|
|
|
|
def _target(operation: dict[str, Any]) -> str:
|
|
target = operation.get("target") or {}
|
|
return str(target.get("instance") or target.get("zone") or "").replace("_", "-")
|
|
|
|
|
|
def _integer(parameters: dict[str, Any], name: str, minimum: int, maximum: int) -> int | None:
|
|
value = parameters.get(name)
|
|
if value is None or value == "" or isinstance(value, bool):
|
|
return None
|
|
try:
|
|
parsed = int(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if str(value).strip() not in {str(parsed), f"+{parsed}"} and not isinstance(value, int):
|
|
return None
|
|
return parsed if minimum <= parsed <= maximum else None
|
|
|
|
|
|
def commands_for_operation(operation: dict[str, Any]) -> list[str]:
|
|
"""Convert one validated CAN_DEV operation to native gateway commands."""
|
|
|
|
capability = str(operation.get("capability") or "")
|
|
parameters = dict(operation.get("parameters") or {})
|
|
target = _target(operation)
|
|
state_value = parameters.get("state", parameters.get("power"))
|
|
on_off = state_value if state_value in {"on", "off"} else None
|
|
|
|
if capability in {"seat.set_driver_heating_native", "seat.set_driver_ventilation_native"}:
|
|
if target != "driver" or not on_off:
|
|
return []
|
|
feature = "heat" if "heating" in capability else "vent"
|
|
if on_off == "off":
|
|
return [f"seat driver {feature} off"]
|
|
level = _integer(parameters, "level", 1, 2)
|
|
state = "high" if level == 1 else "low" if level == 2 else None
|
|
return [f"seat driver {feature} {state}"] if state else []
|
|
|
|
if capability == "seat.set_driver_massage_native":
|
|
if target != "driver":
|
|
return []
|
|
if parameters.get("action") == "off":
|
|
return ["seat driver massage off"]
|
|
if parameters.get("action") == "strength":
|
|
strength = _integer(parameters, "strength", 1, 3)
|
|
return [f"seat driver massage strength {['', 'weak', 'medium', 'strong'][strength]}"] if strength else []
|
|
if parameters.get("action") == "mode" and parameters.get("mode") in {
|
|
"wave", "pulse", "stretch", "slack", "relax",
|
|
}:
|
|
return [f"seat driver massage mode {parameters['mode']}"]
|
|
return []
|
|
|
|
if capability == "seat.set_driver_axis_native":
|
|
directions = {
|
|
"horizontal": {"forward", "backward", "stop"},
|
|
"height": {"up", "down", "stop"},
|
|
"backrest": {"forward", "backward", "stop"},
|
|
"basin": {"up", "down", "stop"},
|
|
}
|
|
axis = parameters.get("axis")
|
|
direction = parameters.get("direction")
|
|
if target != "driver" or direction not in directions.get(axis, set()):
|
|
return []
|
|
if direction == "stop":
|
|
return [f"seat driver {axis} stop"]
|
|
duration = _integer(parameters, "duration_ms", 100, 2000)
|
|
return [f"seat driver {axis} {direction} {duration}"] if duration else []
|
|
|
|
if capability == "seat.set_driver_memory_native":
|
|
action = parameters.get("action")
|
|
position = _integer(parameters, "position", 1, 3)
|
|
return [f"seat driver memory {action} {position}"] if target == "driver" and action in {"recall", "save"} and position else []
|
|
|
|
if capability in {"seat.set_heating", "seat.set_ventilation"}:
|
|
seats = {"driver": "driver", "front-left": "driver"}
|
|
if seats.get(target) != "driver" or not on_off:
|
|
return []
|
|
feature = "heat" if capability.endswith("heating") else "vent"
|
|
if on_off == "off":
|
|
return [f"seat driver {feature} off"]
|
|
level = _integer(parameters, "level", 1, 5)
|
|
return [f"seat driver {feature} {'high' if level and level >= 3 else 'low'}"]
|
|
|
|
if capability in {"climate.set_power", "climate.set_ac"}:
|
|
return [f"ac {on_off}"] if on_off and target in {"vehicle", "cabin"} else []
|
|
if capability == "climate.set_fan_level":
|
|
level = _integer(parameters, "fan_level", 1, 7)
|
|
if level is None:
|
|
level = _integer(parameters, "level", 1, 7)
|
|
return [f"ac fan {level}"] if level is not None else []
|
|
if capability == "climate.set_temperature":
|
|
temperature = _integer(parameters, "temperature_c", 18, 32)
|
|
zones = {
|
|
"driver": ["driver"], "front-left": ["driver"],
|
|
"passenger": ["passenger"], "front-right": ["passenger"],
|
|
"rear": ["rear"], "rear-row": ["rear"], "rear-left": ["rear"],
|
|
"rear-right": ["rear"], "front-row": ["driver", "passenger"],
|
|
"vehicle": ["driver", "passenger"], "cabin": ["driver", "passenger"],
|
|
}.get(target)
|
|
return [f"ac temp {zone} {temperature}" for zone in zones or []] if temperature is not None else []
|
|
if capability == "climate.set_outlet_mode":
|
|
aliases = {
|
|
"face": "face", "face-feet": "face-feet", "face_feet": "face-feet",
|
|
"feet": "feet", "feet-defrost": "feet-defrost", "feet_defrost": "feet-defrost",
|
|
"defrost": "defrost", "face-feet-defrost": "face-feet-defrost",
|
|
"face_feet_defrost": "face-feet-defrost", "face-defrost": "face-defrost",
|
|
"face_defrost": "face-defrost",
|
|
}
|
|
mode = aliases.get(parameters.get("outlet_mode", parameters.get("mode")))
|
|
return [f"ac mode {mode}"] if mode and target in {"vehicle", "cabin"} else []
|
|
if capability == "climate.set_front_defrost" and on_off == "on":
|
|
return ["ac mode defrost"]
|
|
if capability == "climate.set_control_mode" and parameters.get("mode") in {"manual", "auto"}:
|
|
return [f"ac control {parameters['mode']}"]
|
|
|
|
if capability == "door.set_rear_lock":
|
|
state = parameters.get("state")
|
|
return [f"door rear {state}"] if target == "rear" and state in {"lock", "unlock"} else []
|
|
if capability == "window.set_motion":
|
|
motion = parameters.get("motion")
|
|
scopes = {"all", "front", "rear", "front-left", "front-right", "rear-left", "rear-right"}
|
|
return [f"window {target} {motion}"] if target in scopes and motion in {"open", "close", "stop"} else []
|
|
if capability == "window.set_position":
|
|
position = _integer(parameters, "position_pct", 0, 100)
|
|
scopes = {"front-left", "front-right", "rear-left", "rear-right"}
|
|
return [f"window {target} position {position}"] if target in scopes and position is not None else []
|
|
|
|
if capability in {"outlet.set_position_percent", "outlet.set_calibrated_position"}:
|
|
scopes = {"driver-left", "driver-right", "passenger-left", "passenger-right"}
|
|
if capability.endswith("percent"):
|
|
horizontal = _integer(parameters, "horizontal_pct", 0, 100)
|
|
vertical = _integer(parameters, "vertical_pct", 0, 100)
|
|
command_type = "percent"
|
|
else:
|
|
horizontal = _integer(parameters, "horizontal_value", 0, 255)
|
|
vertical = _integer(parameters, "vertical_value", 0, 255)
|
|
command_type = "position"
|
|
return [f"outlet {target} {command_type} {horizontal} {vertical}"] if target in scopes and horizontal is not None and vertical is not None else []
|
|
if capability == "outlet.set_airflow_mode":
|
|
mode, state = parameters.get("mode"), parameters.get("state")
|
|
return [f"outlet mode {target} {mode} {state}"] if target in {"driver", "passenger"} and mode in {"person", "avoid", "free"} and state in {"on", "off"} else []
|
|
if capability == "outlet.set_sweep_mode":
|
|
axis = {"up_down": "up-down", "left_right": "left-right"}.get(parameters.get("axis"))
|
|
action = parameters.get("action")
|
|
return [f"outlet mode {target} sweep {axis} {action}"] if target in {"driver", "passenger"} and axis and action in {"start", "stop"} else []
|
|
if capability == "outlet.set_power":
|
|
state = parameters.get("state")
|
|
return [f"outlet mode {target} {state}"] if target in {"driver", "passenger"} and state in {"open", "close"} else []
|
|
|
|
if capability == "roof.set_shade_position":
|
|
position = _integer(parameters, "shade_position_pct", 0, 100)
|
|
if position is None:
|
|
position = _integer(parameters, "position_pct", 0, 100)
|
|
return [f"shade position {position}"] if position is not None else []
|
|
if capability == "roof.set_shade_motion" and parameters.get("motion") in {"open", "close", "half", "stop"}:
|
|
return [f"shade {parameters['motion']}"]
|
|
|
|
if capability == "interior_light.set_indoor_power":
|
|
return [f"light indoor {on_off}"] if on_off else []
|
|
if capability == "interior_light.set_reading_power":
|
|
return [f"light reading all {on_off}"] if target == "all" and on_off else []
|
|
if capability == "ambient_light.set_power":
|
|
return [f"light ambient {on_off}"] if on_off else []
|
|
if capability == "ambient_light.set_color_index_native":
|
|
color = _integer(parameters, "color_index", 1, 127)
|
|
return [f"light ambient color {color}"] if color is not None else []
|
|
if capability == "ambient_light.set_brightness_level_native":
|
|
level = _integer(parameters, "level", 1, 5)
|
|
return [f"light ambient brightness {level}"] if level is not None else []
|
|
if capability == "ambient_light.set_mode" and parameters.get("mode") in {"on", "off"}:
|
|
return [f"light ambient {parameters['mode']}"]
|
|
return []
|
|
|
|
|
|
def extract_native_commands(result: dict[str, Any]) -> tuple[list[str], list[str]]:
|
|
"""Return unique commands and explicit labels for unsupported operations."""
|
|
|
|
operations: list[dict[str, Any]] = []
|
|
for config in (result.get("function_configs") or {}).values():
|
|
operations.extend((config or {}).get("spec", {}).get("operations", []))
|
|
for request in result.get("deferred_requests") or []:
|
|
parameters = request.get("parameters") or {}
|
|
if isinstance(parameters, list):
|
|
parameters = {item.get("name"): item.get("value") for item in parameters if isinstance(item, dict)}
|
|
operations.append({
|
|
"capability": request.get("capability"),
|
|
"target": {"zone": request.get("target_scope")},
|
|
"parameters": parameters,
|
|
})
|
|
|
|
commands: list[str] = []
|
|
unsupported: list[str] = []
|
|
for operation in operations:
|
|
converted = commands_for_operation(operation)
|
|
if converted:
|
|
commands.extend(converted)
|
|
else:
|
|
unsupported.append(str(operation.get("capability") or "unknown"))
|
|
return list(dict.fromkeys(commands)), list(dict.fromkeys(unsupported))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class NativeCanSettings:
|
|
endpoint: str
|
|
execution_mode: str
|
|
approval_token: str
|
|
timeout: float
|
|
|
|
@classmethod
|
|
def from_env(cls) -> "NativeCanSettings":
|
|
mode = os.environ.get("COGNITIVE_AGENT_EXECUTION_MODE", "dry_run").strip().lower()
|
|
if mode not in {"dry_run", "hardware"}:
|
|
mode = "dry_run"
|
|
return cls(
|
|
endpoint=os.environ.get("COGNITIVE_AGENT_NATIVE_CAN_URL", "http://127.0.0.1:8765").rstrip("/"),
|
|
execution_mode=mode,
|
|
approval_token=os.environ.get("COGNITIVE_AGENT_NATIVE_CAN_APPROVAL_TOKEN", "").strip(),
|
|
timeout=float(os.environ.get("COGNITIVE_AGENT_NATIVE_CAN_TIMEOUT", "20")),
|
|
)
|
|
|
|
|
|
class NativeCanClient:
|
|
def __init__(self, settings: NativeCanSettings | None = None):
|
|
self.settings = settings or NativeCanSettings.from_env()
|
|
|
|
def _request(self, path: str, *, body: str | None = None, token: bool = False, timeout: float | None = None) -> tuple[int, dict[str, Any]]:
|
|
headers = {"Accept": "application/json"}
|
|
data = None
|
|
if body is not None:
|
|
data = body.encode("utf-8")
|
|
headers["Content-Type"] = "text/plain; charset=utf-8"
|
|
if token:
|
|
headers["X-CAN-Approval-Token"] = self.settings.approval_token
|
|
request = Request(self.settings.endpoint + path, data=data, headers=headers, method="POST" if data is not None else "GET")
|
|
try:
|
|
with urlopen(request, timeout=timeout or self.settings.timeout) as response: # noqa: S310 - operator configured endpoint
|
|
raw, status = response.read(), int(response.status)
|
|
except HTTPError as exc:
|
|
raw, status = exc.read(), int(exc.code)
|
|
except (URLError, TimeoutError, OSError) as exc:
|
|
raise NativeCanError(f"无法连接原生 CAN 网关：{urlparse(self.settings.endpoint).hostname or self.settings.endpoint}") from exc
|
|
try:
|
|
payload = json.loads(raw.decode("utf-8")) if raw else {}
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise NativeCanError("原生 CAN 网关返回了无法识别的数据。") from exc
|
|
if not isinstance(payload, dict):
|
|
raise NativeCanError("原生 CAN 网关未返回 JSON 对象。")
|
|
return status, payload
|
|
|
|
def status(self) -> dict[str, Any]:
|
|
try:
|
|
status, runtime = self._request("/api/can/runtime", timeout=0.8)
|
|
connected = 200 <= status < 300 and runtime.get("running") is not False
|
|
return {
|
|
"connected": connected,
|
|
"endpoint": self.settings.endpoint,
|
|
"execution_mode": self.settings.execution_mode,
|
|
"gateway_mode": runtime.get("mode", "unknown"),
|
|
"control_ready": bool(connected and runtime.get("mode") == "control" and runtime.get("control_token_configured")),
|
|
"adapter": runtime.get("adapter", {}),
|
|
"latest_state": runtime,
|
|
"message": "原生 CAN 网关在线" if connected else "原生 CAN 网关异常",
|
|
}
|
|
except NativeCanError as exc:
|
|
return {
|
|
"connected": False,
|
|
"endpoint": self.settings.endpoint,
|
|
"execution_mode": self.settings.execution_mode,
|
|
"gateway_mode": "offline",
|
|
"control_ready": False,
|
|
"message": str(exc),
|
|
}
|
|
|
|
def process(self, result: dict[str, Any]) -> dict[str, Any]:
|
|
commands, unsupported = extract_native_commands(result)
|
|
report: dict[str, Any] = {
|
|
"mode": "native_can" if self.settings.execution_mode == "hardware" else "dry_run",
|
|
"status": "blocked",
|
|
"commands": [],
|
|
"skipped": unsupported,
|
|
"target_state_confirmed": False,
|
|
"physical_state_confirmed": False,
|
|
"message": "没有可映射的受信任 CAN 命令。",
|
|
}
|
|
if not commands:
|
|
return report
|
|
|
|
validated: list[str] = []
|
|
for command in commands:
|
|
status, payload = self._request("/api/can/validate", body=command)
|
|
if not 200 <= status < 300 or payload.get("valid") is not True or payload.get("transmitted") is not False:
|
|
report["message"] = f"CAN 只读校验失败，未发送任何控制帧：{payload.get('message') or command}"
|
|
return report
|
|
validated.append(str(payload.get("command") or command))
|
|
|
|
if self.settings.execution_mode != "hardware":
|
|
report.update({
|
|
"status": "validated",
|
|
"commands": [{"command": command, "status": "validated"} for command in validated],
|
|
"message": f"{len(validated)} 条车控命令已通过只读校验；当前为安全演练模式，未发送 CAN。",
|
|
})
|
|
return report
|
|
if len(self.settings.approval_token) < 6:
|
|
report["message"] = "实车模式缺少至少 6 字符的原生 CAN 授权令牌。"
|
|
return report
|
|
|
|
report["status"] = "running"
|
|
target_state_confirmed = True
|
|
for command in validated:
|
|
status, payload = self._request("/api/can/command", body=command, token=True)
|
|
accepted = 200 <= status < 300 and payload.get("accepted") is True
|
|
execution = payload.get("execution") or {}
|
|
report["commands"].append({
|
|
"command": str(payload.get("command") or command),
|
|
"status": "confirmed" if execution.get("feedback_matched") else "accepted" if accepted else "failed",
|
|
"message": payload.get("message") or payload.get("meaning"),
|
|
"execution": execution,
|
|
})
|
|
if not accepted:
|
|
report["status"] = "failed"
|
|
report["message"] = str(payload.get("message") or "CAN 命令失败，后续动作已停止。")
|
|
return report
|
|
target_state_confirmed = (
|
|
target_state_confirmed
|
|
and execution.get("feedback_matched") is True
|
|
)
|
|
|
|
report["status"] = "confirmed" if target_state_confirmed else "accepted"
|
|
report["target_state_confirmed"] = target_state_confirmed
|
|
# CAN feedback proves that a target state appeared on the bus. It does
|
|
# not independently prove that the physical mechanism moved; that
|
|
# requires a dedicated physical sensor or operator/HIL observation.
|
|
report["physical_state_confirmed"] = False
|
|
report["message"] = (
|
|
f"{len(validated)} 条 CAN 命令均收到目标状态反馈（不等同于实体机构确认）。"
|
|
if target_state_confirmed
|
|
else f"{len(validated)} 条 CAN 命令已由适配器提交；部分动作尚无目标状态反馈，请核对实体设备。"
|
|
)
|
|
try:
|
|
_, report["latest_state"] = self._request("/api/can/status", timeout=1.5)
|
|
except NativeCanError:
|
|
pass
|
|
return report
|