Miayui
347 lines
16 KiB
Python
347 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Run a bounded live-vehicle test through the self-expand frontend.
|
|
|
|
This suite intentionally has an explicit confirmation switch because it sends
|
|
real CAN frames. It snapshots the decoded vehicle state, tests one bounded
|
|
change per supported family, restores the original state immediately, and
|
|
records whether the frontend reported target-state feedback. Physical motion
|
|
is never inferred from CAN feedback alone.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
LIVE_CONFIRMATION = "YES-I-UNDERSTAND"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class HilCase:
|
|
case_id: str
|
|
action: str
|
|
restore: str | None
|
|
expected_status: str = "confirmed"
|
|
|
|
|
|
def request_json(url: str, *, payload: dict[str, Any] | None = None, timeout: float = 45) -> tuple[int, dict[str, Any]]:
|
|
data = None
|
|
headers = {"Accept": "application/json"}
|
|
if payload is not None:
|
|
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
headers["Content-Type"] = "application/json; charset=utf-8"
|
|
request = Request(url, data=data, headers=headers, method="POST" if data is not None else "GET")
|
|
try:
|
|
with urlopen(request, timeout=timeout) as response: # noqa: S310 - operator-selected local 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 RuntimeError(f"cannot reach {url}: {exc}") from exc
|
|
decoded = json.loads(raw.decode("utf-8")) if raw else {}
|
|
if not isinstance(decoded, dict):
|
|
raise RuntimeError(f"non-object JSON response from {url}")
|
|
return status, decoded
|
|
|
|
|
|
def state(runtime: dict[str, Any], frame_id: str) -> dict[str, Any]:
|
|
return dict((runtime.get("states") or {}).get(frame_id) or {})
|
|
|
|
|
|
def power_utterance(prefix: str, raw: int) -> str:
|
|
return f"{'打开' if raw == 2 else '关闭'}{prefix}"
|
|
|
|
|
|
def thermal_utterance(feature: str, raw: int) -> str:
|
|
if raw == 1:
|
|
return f"关闭主驾座椅{feature}"
|
|
if raw == 3:
|
|
return f"主驾座椅{feature}高档"
|
|
return f"打开主驾座椅{feature}"
|
|
|
|
|
|
def build_cases(runtime: dict[str, Any], include_outlets: bool) -> list[HilCase]:
|
|
climate = state(runtime, "0x404")
|
|
seat = state(runtime, "0x4FA")
|
|
light = state(runtime, "0x42E")
|
|
shade = state(runtime, "0x41A")
|
|
window = state(runtime, "0x388")
|
|
door = state(runtime, "0x407")
|
|
required = {"climate": climate, "seat": seat, "light": light, "shade": shade, "window": window, "door": door}
|
|
missing = [name for name, value in required.items() if not value]
|
|
if missing:
|
|
raise RuntimeError(f"missing live state needed for restoration: {', '.join(missing)}")
|
|
|
|
positions = window.get("positions") or []
|
|
if len(positions) != 4 or any(value is None for value in positions):
|
|
raise RuntimeError("window positions are not available for safe restoration")
|
|
front_left_restore = "关闭全部车窗" if int(positions[0]) <= 3 else f"把左前车窗打开到{int(positions[0])}%"
|
|
shade_position = int(shade.get("position"))
|
|
if shade_position <= 2:
|
|
shade_restore = "关闭遮阳帘"
|
|
elif shade_position >= 98:
|
|
shade_restore = "打开遮阳帘"
|
|
elif 47 <= shade_position <= 53:
|
|
shade_restore = "把遮阳帘打开一半"
|
|
else:
|
|
raise RuntimeError("shade is not at a restorable 0/50/100 position")
|
|
|
|
power_restore = "打开空调" if int(climate.get("power")) == 1 else "关闭空调"
|
|
control_restore = "把空调设置为手动模式" if int(climate.get("control")) == 1 else "把空调设置为自动模式"
|
|
ambient_raw = int(light.get("front_brightness"))
|
|
ambient_restore = "关闭车内氛围灯" if ambient_raw == 1 else "打开车内氛围灯"
|
|
indoor_restore = power_utterance("车内室内灯", int(light.get("indoor")))
|
|
door_restore = "解锁后排门锁" if int(door.get("rear_left")) == 1 else "锁上后排门锁"
|
|
|
|
cases = [
|
|
HilCase("climate-power", "关闭空调" if int(climate.get("power")) == 1 else "打开空调", power_restore),
|
|
HilCase("climate-temperature", "把空调温度调到24度", f"把空调温度调到{int(climate.get('driver_temp'))}度"),
|
|
HilCase("climate-fan", "把空调风量调到3档", f"把空调风量调到{int(climate.get('fan'))}档"),
|
|
HilCase("climate-control", "把空调设置为自动模式" if int(climate.get("control")) != 2 else "把空调设置为手动模式", control_restore),
|
|
HilCase("seat-ventilation", "关闭主驾座椅通风" if int(seat.get("vent")) != 1 else "打开主驾座椅通风", thermal_utterance("通风", int(seat.get("vent")))),
|
|
HilCase("seat-heating", "关闭主驾座椅加热" if int(seat.get("heat")) != 1 else "打开主驾座椅加热", thermal_utterance("加热", int(seat.get("heat")))),
|
|
HilCase("light-indoor", "关闭车内室内灯" if int(light.get("indoor")) == 2 else "打开车内室内灯", indoor_restore),
|
|
HilCase("light-ambient", "关闭车内氛围灯" if ambient_raw != 1 else "打开车内氛围灯", ambient_restore),
|
|
# The current vehicle acknowledges the trusted rear-lock request on the
|
|
# adapter but may inhibit the state transition. The UI must preserve
|
|
# that distinction as "accepted", not fabricate target confirmation.
|
|
HilCase("door-rear-lock", "锁上后排门锁" if int(door.get("rear_left")) == 1 else "解锁后排门锁", door_restore, "accepted"),
|
|
HilCase("window-front-left", "把左前车窗打开到50%", front_left_restore),
|
|
HilCase("shade", "把遮阳帘打开一半", shade_restore),
|
|
]
|
|
if include_outlets:
|
|
cases.extend(
|
|
HilCase(f"outlet-{zone}", f"把{name}风口调到中间位置", None, "accepted")
|
|
for zone, name in (
|
|
("driver-left", "主驾左"),
|
|
("driver-right", "主驾右"),
|
|
("passenger-left", "副驾左"),
|
|
("passenger-right", "副驾右"),
|
|
)
|
|
)
|
|
return cases
|
|
|
|
|
|
def final_restore_commands(runtime: dict[str, Any]) -> list[str]:
|
|
climate = state(runtime, "0x404")
|
|
seat = state(runtime, "0x4FA")
|
|
light = state(runtime, "0x42E")
|
|
shade = state(runtime, "0x41A")
|
|
window = state(runtime, "0x388")
|
|
door = state(runtime, "0x407")
|
|
climate_modes = {
|
|
1: "空调吹面",
|
|
2: "空调吹面吹脚",
|
|
3: "空调吹脚",
|
|
4: "空调吹脚并除霜",
|
|
5: "空调除霜",
|
|
6: "空调吹面吹脚并除霜",
|
|
7: "空调吹面并除霜",
|
|
}
|
|
commands = [
|
|
"打开空调" if int(climate["power"]) == 1 else "关闭空调",
|
|
f"把空调温度调到{int(climate['driver_temp'])}度",
|
|
f"把空调风量调到{int(climate['fan'])}档",
|
|
"把空调设置为手动模式" if int(climate["control"]) == 1 else "把空调设置为自动模式",
|
|
climate_modes[int(climate["mode"])],
|
|
thermal_utterance("通风", int(seat["vent"])),
|
|
thermal_utterance("加热", int(seat["heat"])),
|
|
power_utterance("车内室内灯", int(light["indoor"])),
|
|
"关闭车内氛围灯" if int(light["front_brightness"]) == 1 else "打开车内氛围灯",
|
|
"解锁后排门锁" if int(door["rear_left"]) == 1 else "锁上后排门锁",
|
|
"关闭遮阳帘" if int(shade["position"]) <= 2 else "打开遮阳帘" if int(shade["position"]) >= 98 else "把遮阳帘打开一半",
|
|
]
|
|
if int(light["front_brightness"]) > 1:
|
|
commands.append(f"把氛围灯亮度调到{int(light['front_brightness']) - 1}档")
|
|
window_names = ("左前", "左后", "右前", "右后")
|
|
for name, position in zip(window_names, window["positions"]):
|
|
commands.append("关闭全部车窗" if all(int(value) <= 3 for value in window["positions"]) else f"把{name}车窗打开到{int(position)}%")
|
|
if commands[-1] == "关闭全部车窗":
|
|
break
|
|
return commands
|
|
|
|
|
|
def restored_state_matches(before: dict[str, Any], after: dict[str, Any]) -> tuple[bool, dict[str, Any]]:
|
|
fields = {
|
|
"0x404": ("power", "control", "mode", "fan", "driver_temp", "passenger_temp"),
|
|
"0x4FA": ("vent", "heat"),
|
|
"0x388": ("positions", "closed"),
|
|
"0x42E": ("front_brightness", "rear_brightness", "front_color", "rear_color", "indoor"),
|
|
"0x41A": ("position",),
|
|
"0x407": ("rear_left", "rear_right", "logical"),
|
|
}
|
|
differences: dict[str, Any] = {}
|
|
for frame_id, names in fields.items():
|
|
before_state, after_state = state(before, frame_id), state(after, frame_id)
|
|
for name in names:
|
|
if before_state.get(name) != after_state.get(name):
|
|
differences[f"{frame_id}.{name}"] = {
|
|
"before": before_state.get(name),
|
|
"after": after_state.get(name),
|
|
}
|
|
return not differences, differences
|
|
|
|
|
|
def execute(frontend: str, vehicle_id: str, utterance: str, timeout: float) -> tuple[int, dict[str, Any]]:
|
|
return request_json(
|
|
f"{frontend}/api/vehicle/execute",
|
|
payload={"command": utterance, "vehicle_id": vehicle_id},
|
|
timeout=timeout,
|
|
)
|
|
|
|
|
|
def summarize_response(status: int, response: dict[str, Any]) -> dict[str, Any]:
|
|
report = response.get("native_can_execution") or {}
|
|
return {
|
|
"http_status": status,
|
|
"success": response.get("success"),
|
|
"report_status": report.get("status"),
|
|
"target_state_confirmed": report.get("target_state_confirmed"),
|
|
"physical_state_confirmed": report.get("physical_state_confirmed"),
|
|
"commands": [item.get("command") for item in report.get("commands", []) if isinstance(item, dict)],
|
|
"message": report.get("message"),
|
|
}
|
|
|
|
|
|
def run(args: argparse.Namespace) -> dict[str, Any]:
|
|
if args.confirm_live_control != LIVE_CONFIRMATION:
|
|
raise RuntimeError(f"live control requires --confirm-live-control {LIVE_CONFIRMATION}")
|
|
frontend = args.frontend_url.rstrip("/")
|
|
native = args.native_url.rstrip("/")
|
|
started = time.time()
|
|
_, health = request_json(f"{frontend}/api/health", timeout=5)
|
|
vehicle = health.get("vehicle") or {}
|
|
native_health = vehicle.get("native_can") or {}
|
|
if not (
|
|
health.get("success") is True
|
|
and vehicle.get("execution_mode") == "hardware"
|
|
and vehicle.get("control_ready") is True
|
|
and native_health.get("gateway_mode") == "control"
|
|
):
|
|
raise RuntimeError("frontend is not connected to a control-ready hardware gateway")
|
|
|
|
_, before = request_json(f"{native}/api/can/runtime", timeout=5)
|
|
gear = state(before, "0x342").get("gear")
|
|
speed = state(before, "0x4EB").get("speed_kmh")
|
|
if gear != "P" or (speed is not None and float(speed) > 0.1):
|
|
raise RuntimeError(f"live motion tests require gear P and zero speed; gear={gear}, speed={speed}")
|
|
before_tx = int(before.get("tx_count", -1))
|
|
cases = build_cases(before, args.include_outlets)
|
|
results: list[dict[str, Any]] = []
|
|
|
|
for case in cases:
|
|
action_status, action_response = execute(frontend, args.vehicle_id, case.action, args.action_timeout)
|
|
action_summary = summarize_response(action_status, action_response)
|
|
action_ok = (
|
|
action_summary["http_status"] < 300
|
|
and action_summary["success"] is True
|
|
and action_summary["report_status"] == case.expected_status
|
|
and action_summary["physical_state_confirmed"] is False
|
|
and (
|
|
action_summary["target_state_confirmed"] is True
|
|
if case.expected_status == "confirmed"
|
|
else action_summary["target_state_confirmed"] is False
|
|
)
|
|
)
|
|
restore_summary: dict[str, Any] | None = None
|
|
restore_ok = case.restore is None
|
|
if case.restore is not None:
|
|
restore_status, restore_response = execute(frontend, args.vehicle_id, case.restore, args.action_timeout)
|
|
restore_summary = summarize_response(restore_status, restore_response)
|
|
restore_ok = (
|
|
restore_summary["http_status"] < 300
|
|
and restore_summary["success"] is True
|
|
and restore_summary["report_status"] == "confirmed"
|
|
and restore_summary["physical_state_confirmed"] is False
|
|
)
|
|
results.append(
|
|
{
|
|
"id": case.case_id,
|
|
"passed": action_ok and restore_ok,
|
|
"action": case.action,
|
|
"action_result": action_summary,
|
|
"restore": case.restore,
|
|
"restore_result": restore_summary,
|
|
}
|
|
)
|
|
time.sleep(args.inter_case_delay)
|
|
|
|
restore_results = []
|
|
for utterance in final_restore_commands(before):
|
|
restore_status, restore_response = execute(frontend, args.vehicle_id, utterance, args.action_timeout)
|
|
restore_results.append({"utterance": utterance, **summarize_response(restore_status, restore_response)})
|
|
time.sleep(0.5)
|
|
|
|
_, after = request_json(f"{native}/api/can/runtime", timeout=5)
|
|
after_tx = int(after.get("tx_count", -1))
|
|
results.append(
|
|
{
|
|
"id": "live-transmit-proof",
|
|
"passed": after_tx > before_tx >= 0,
|
|
"before_tx": before_tx,
|
|
"after_tx": after_tx,
|
|
}
|
|
)
|
|
state_restored, differences = restored_state_matches(before, after)
|
|
results.append(
|
|
{
|
|
"id": "final-state-restored",
|
|
"passed": state_restored and all(item["success"] is True for item in restore_results),
|
|
"differences": differences,
|
|
"restore_results": restore_results,
|
|
}
|
|
)
|
|
passed_count = sum(item["passed"] for item in results)
|
|
return {
|
|
"suite": "vehicle_frontend_hil",
|
|
"passed": passed_count == len(results),
|
|
"summary": {"passed": passed_count, "failed": len(results) - passed_count, "total": len(results)},
|
|
"duration_ms": round((time.time() - started) * 1000),
|
|
"frontend_url": frontend,
|
|
"native_url": native,
|
|
"baseline": {
|
|
"tx_count": before_tx,
|
|
"states": before.get("states", {}),
|
|
},
|
|
"final": {
|
|
"tx_count": after_tx,
|
|
"states": after.get("states", {}),
|
|
},
|
|
"results": results,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--frontend-url", default="http://127.0.0.1:8010")
|
|
parser.add_argument("--native-url", default="http://127.0.0.1:8765")
|
|
parser.add_argument("--vehicle-id", default="denza_n7")
|
|
parser.add_argument("--confirm-live-control", default="")
|
|
parser.add_argument("--include-outlets", action="store_true")
|
|
parser.add_argument("--action-timeout", type=float, default=45)
|
|
parser.add_argument("--inter-case-delay", type=float, default=0.3)
|
|
parser.add_argument("--output", type=Path)
|
|
args = parser.parse_args()
|
|
try:
|
|
report = run(args)
|
|
except Exception as exc: # noqa: BLE001 - command-line diagnostic boundary
|
|
report = {"suite": "vehicle_frontend_hil", "passed": False, "fatal_error": str(exc)}
|
|
rendered = json.dumps(report, ensure_ascii=False, indent=2)
|
|
if args.output:
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(rendered + "\n", encoding="utf-8")
|
|
print(rendered)
|
|
return 0 if report.get("passed") else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|