feat: integrate NFC infrared peripherals and orchestration

This commit is contained in:
vee
2026-08-25 11:45:45 +08:00
parent 9ac8fdface
commit 2529b4896a
19 changed files with 3818 additions and 117 deletions
+179 -5
View File
@@ -347,11 +347,11 @@ operations:
) as response:
console_styles = response.read().decode("utf-8")
with self.opener.open(
f"{self.base_url}/self-extension.js?v=sync-layout-26", timeout=3
f"{self.base_url}/self-extension.js?v=nfc-toy-33", timeout=3
) as response:
extension_script = response.read().decode("utf-8")
with self.opener.open(
f"{self.base_url}/self-extension.css?v=sync-layout-26", timeout=3
f"{self.base_url}/self-extension.css?v=nfc-toy-33", timeout=3
) as response:
extension_styles = response.read().decode("utf-8").replace("\r\n", "\n")
with self.opener.open(
@@ -362,6 +362,10 @@ operations:
f"{self.base_url}/api/extension/health", timeout=3
) as response:
extension_health = json.loads(response.read().decode("utf-8"))
with self.opener.open(
f"{self.base_url}/api/extension/nfc/status", timeout=3
) as response:
nfc_status = json.loads(response.read().decode("utf-8"))
with self.opener.open(f"{self.base_url}/api/runtime", timeout=3) as response:
runtime = json.loads(response.read().decode("utf-8"))
@@ -387,8 +391,8 @@ operations:
self.assertIn('<body class="console-v2">', html)
self.assertIn('id="v2BootScreen"', html)
self.assertIn('正在加载车机控制与实时状态', html)
self.assertIn('src="/self-extension.js?v=sync-layout-26"', html)
self.assertIn('href="/self-extension.css?v=sync-layout-26"', html)
self.assertIn('src="/self-extension.js?v=nfc-toy-33"', html)
self.assertIn('href="/self-extension.css?v=nfc-toy-33"', html)
self.assertNotIn("self-calibration", html)
self.assertNotIn("self-repair", html)
self.assertNotIn("self-evolution", html)
@@ -594,6 +598,12 @@ operations:
self.assertIn('grid-template-columns: repeat(3, minmax(0, 1fr))', extension_styles)
self.assertTrue(extension_health["success"])
self.assertIn(extension_health["transport"], {"tcp", "mqtt", "simulation"})
self.assertIn("nfc", extension_health)
self.assertTrue(nfc_status["success"])
self.assertIn("reader", nfc_status)
self.assertIn("card", nfc_status)
self.assertIn("legacy_profiles", nfc_status)
self.assertIn("legacy_bridge_addresses", nfc_status)
self.assertIn("function emphasizeReferenceFrontLampMeshes", app_script)
self.assertIn("function createReferenceFrontLampHighlights", app_script)
self.assertIn("referenceFrontLampMaterials", app_script)
@@ -612,7 +622,7 @@ operations:
self.assertIn('client_type: "web_console"', app_script)
self.assertIn('canvas.addEventListener("webglcontextlost"', app_script)
self.assertIn('"cognitive-agent:module-change"', app_script)
self.assertIn('const CACHE_VERSION = "cognitive-agent-pad-v204"', service_worker)
self.assertIn('const CACHE_VERSION = "cognitive-agent-pad-v210"', service_worker)
self.assertIn('"/assets/pets/self-extension-cat.png?v=3"', service_worker)
self.assertIn(".then(() => self.skipWaiting())", service_worker)
self.assertNotIn("const cached = await caches.match(request)", service_worker)
@@ -1009,6 +1019,170 @@ operations:
self.assertEqual(result["transport"], "simulation")
self.assertFalse(result["physical_state_confirmed"])
def test_main_request_routes_live_nfc_peripheral_to_gscu(self) -> None:
original_extension = SeatDemoHandler.extension
original_nfc = SeatDemoHandler.nfc
SeatDemoHandler.extension = SelfExtensionService(SimulationGateway())
SeatDemoHandler.nfc = SimpleNamespace(
snapshot=lambda: {
"revision": 1,
"reader": {"online": True},
"card": {"action": "present", "uid": "1A53DD95"},
}
)
catalog_request = urllib.request.Request(
f"{self.base_url}/api/extension/display-sync",
method="POST",
headers={"Content-Type": "application/json"},
data=json.dumps(
{
"peripherals": [
{
"id": "cap-third-party-light-on",
"status": "learned",
"device_id": "third-party-light",
"device_name": "第三方氛围灯",
"card_uid": "1A53DD95",
"action": "开启",
"aliases": ["开灯"],
"slot": 9,
}
]
},
ensure_ascii=False,
).encode("utf-8"),
)
request = urllib.request.Request(
f"{self.base_url}/api/requests",
method="POST",
headers={"Content-Type": "application/json"},
data=json.dumps(
{
"utterance": "打开第三方氛围灯",
"provider": "local",
"conversation_id": "conv-main-nfc-extension",
"execution_mode": "hardware",
"client_type": "web_console",
},
ensure_ascii=False,
).encode("utf-8"),
)
try:
with self.opener.open(catalog_request, timeout=3) as response:
self.assertEqual(response.status, 200)
with patch.object(
self.service,
"process",
side_effect=AssertionError("vehicle orchestration must not run"),
):
with self.opener.open(request, timeout=3) as response:
result = json.loads(response.read().decode("utf-8"))
finally:
SeatDemoHandler.extension = original_extension
SeatDemoHandler.nfc = original_nfc
self.assertEqual(result["status"], "succeeded")
self.assertEqual(result["orchestration_route"], "nfc_peripheral")
self.assertEqual(result["peripheral"]["name"], "第三方氛围灯")
self.assertEqual(result["peripheral"]["slot"], 9)
operation = result["execution_report"]["spec"]["results"][0]
self.assertEqual(operation["domain"], "self_extension")
self.assertTrue(operation["telemetry"]["gateway_acknowledged"])
def test_main_request_executes_gpt_nfc_sequence_without_vehicle_can(self) -> None:
original_extension = SeatDemoHandler.extension
original_nfc = SeatDemoHandler.nfc
extension = SelfExtensionService(SimulationGateway())
SeatDemoHandler.extension = extension
SeatDemoHandler.nfc = SimpleNamespace(
snapshot=lambda: {
"revision": 1,
"reader": {"online": True},
"card": {"action": "present", "uid": "1A53DD95"},
}
)
capabilities = [
{
"id": capability_id,
"status": "learned",
"device_id": "third-party-light",
"device_name": "第三方氛围灯",
"card_uid": "1A53DD95",
"action": action,
"slot": slot,
}
for capability_id, action, slot in (
("on", "开启灯光", 1),
("red", "红色灯光", 2),
("off", "关闭灯光", 3),
)
]
catalog_request = urllib.request.Request(
f"{self.base_url}/api/extension/display-sync",
method="POST",
headers={"Content-Type": "application/json"},
data=json.dumps(
{"peripherals": capabilities}, ensure_ascii=False
).encode("utf-8"),
)
utterance = "打开第三方氛围灯,红灯闪烁5s后,最后关闭第三方氛围灯"
request = urllib.request.Request(
f"{self.base_url}/api/requests",
method="POST",
headers={"Content-Type": "application/json"},
data=json.dumps(
{
"utterance": utterance,
"provider": "local",
"conversation_id": "conv-main-nfc-sequence",
"execution_mode": "hardware",
"client_type": "web_console",
},
ensure_ascii=False,
).encode("utf-8"),
)
plan_result = {
"model": "gpt-5.6-sol",
"summary": "按顺序切换第三方氛围灯",
"needs_confirmation": False,
"steps": [
{"kind": "peripheral", "capability_id": "on"},
{"kind": "peripheral", "capability_id": "red"},
{"kind": "delay", "delay_ms": 5000},
{"kind": "peripheral", "capability_id": "off"},
],
}
try:
with self.opener.open(catalog_request, timeout=3) as response:
self.assertEqual(response.status, 200)
with patch.object(
self.service,
"process",
side_effect=AssertionError("vehicle orchestration must not run"),
), patch.object(
extension, "plan", return_value=plan_result
) as planner, patch(
"cognitive_agent.application.self_extension.time.sleep"
):
with self.opener.open(request, timeout=3) as response:
result = json.loads(response.read().decode("utf-8"))
finally:
SeatDemoHandler.extension = original_extension
SeatDemoHandler.nfc = original_nfc
self.assertEqual(result["status"], "succeeded")
self.assertEqual(
result["orchestration_route"], "nfc_peripheral_sequence"
)
self.assertEqual(result["sequence"]["step_count"], 4)
self.assertEqual(result["sequence"]["action_count"], 3)
self.assertEqual(result["sequence"]["emit_count"], 3)
self.assertEqual(result["sequence"]["steady_blink_fallbacks"], 1)
self.assertEqual(result["sequence"]["total_delay_ms"], 5000)
self.assertEqual(len(result["operation_results"]), 4)
planner.assert_called_once()
self.assertEqual(planner.call_args.args[0]["utterance"], utterance)
def test_post_request_returns_complete_dry_run(self) -> None:
request = urllib.request.Request(
f"{self.base_url}/api/requests",