vee/Cognitive_Agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

完善模式卡片、NFC及多设备控制流程

Browse Source
This commit is contained in:
minghanchen
2026-09-01 16:18:57 +08:00
parent ca26259e63
commit cbf230182d
15 changed files with 1092 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/e2e/test_seat_web_api.py
+250 -8
View File
@@ -289,6 +289,247 @@ operations:
self.service.mode_profiles = original_registry
self.service.artifact_root = original_artifact_root
def test_office_mode_generation_and_card_execution_reuse_saved_flow(self) -> None:
original_registry = self.service.mode_profiles
original_artifact_root = self.service.artifact_root
original_extension = SeatDemoHandler.extension
original_nfc = SeatDemoHandler.nfc
instruction = (
"空调温度设置为25度，风速设置为1档，"
"启动闪烁灯，颜色设置为红色"
)
with tempfile.TemporaryDirectory() as directory:
base = Path(directory)
preset_root = base / "presets"
user_root = base / "users"
preset = preset_root / "vehicles" / "denza_n7" / "office.yaml"
preset.parent.mkdir(parents=True)
preset.write_text(
"""
id: office
name: 办公模式
icon: briefcase
scope:
vehicle_ids: [denza_n7]
operations:
- domain: climate
capability: climate.set_temperature
target_scope: driver
parameters:
temperature_c: 24
""".strip()
+ "\n",
encoding="utf-8",
)
extension = SelfExtensionService(
SimulationGateway(),
catalog_db_path=base / "office-mode-ir.sqlite3",
)
extension.sync_peripheral_catalog(
[
{
"id": "ir-slot:tony-light-001:1",
"status": "learned",
"enabled": True,
"device_id": "tony-light-001",
"device_name": "闪灯玩具",
"object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
"card_uid": "2CC8B103",
"action": "打开",
"action_names": ["打开", "开启", "启动"],
"risk": "low",
"slot": 1,
},
{
"id": "ir-slot:tony-light-001:3",
"status": "learned",
"enabled": True,
"device_id": "tony-light-001",
"device_name": "闪灯玩具",
"object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
"card_uid": "2CC8B103",
"action": "切换到红色",
"action_names": [
"切换到红色",
"切换为红色",
"设置为红色",
],
"risk": "low",
"slot": 3,
},
]
)
self.service.mode_profiles = ModeProfileRegistry(
preset_root,
user_root=user_root,
)
self.service.artifact_root = base / "runs"
SeatDemoHandler.extension = extension
SeatDemoHandler.nfc = SimpleNamespace(
snapshot=lambda: {
"revision": 1,
"reader": {"online": True},
"card": {"action": "present", "uid": "2CC8B103"},
}
)
try:
generate_request = urllib.request.Request(
f"{self.base_url}/api/requests/preview",
method="POST",
headers={"Content-Type": "application/json"},
data=json.dumps(
{
"utterance": instruction,
"provider": "local",
"conversation_id": "conv-office-mode-generate",
"vehicle_id": "denza_n7",
"execution_mode": "dry_run",
"control_transport": "wired_can",
"client_type": "mode_editor_compile",
},
ensure_ascii=False,
).encode("utf-8"),
)
with self.opener.open(generate_request, timeout=3) as response:
generated = json.loads(response.read().decode("utf-8"))
save_request = urllib.request.Request(
f"{self.base_url}/api/mode-profiles/office",
method="POST",
headers={"Content-Type": "application/json"},
data=json.dumps(
{
"name": "办公模式",
"instruction": instruction,
"source_request_id": generated["request_id"],
"vehicle_id": "denza_n7",
},
ensure_ascii=False,
).encode("utf-8"),
)
with self.opener.open(save_request, timeout=3) as response:
saved = json.loads(response.read().decode("utf-8"))
execute_request = urllib.request.Request(
f"{self.base_url}/api/mode-profiles/office/execute",
method="POST",
headers={"Content-Type": "application/json"},
data=json.dumps(
{
"conversation_id": "conv-office-mode-execute",
"vehicle_id": "denza_n7",
"execution_mode": "dry_run",
"control_transport": "wired_can",
"client_type": "web_console_mode_card",
},
ensure_ascii=False,
).encode("utf-8"),
)
with patch.object(
self.service,
"_interpret",
side_effect=AssertionError(
"saved mode card must not reinterpret natural language"
),
), self.opener.open(execute_request, timeout=3) as response:
executed = json.loads(response.read().decode("utf-8"))
finally:
self.service.mode_profiles = original_registry
self.service.artifact_root = original_artifact_root
SeatDemoHandler.extension = original_extension
SeatDemoHandler.nfc = original_nfc
generated_commands = [
(
command.get("signals", {}).get("debug_command")
or (
f"infrared emit {command.get('signals', {}).get('slot')}"
if command.get("message") == "infrared.emit"
else None
)
)
for plan in generated.get("command_plans", {}).values()
for command in plan.get("spec", {}).get("commands", [])
]
self.assertEqual(
generated_commands,
[
"ac temp driver 25",
"ac fan 1",
"infrared emit 1",
"infrared emit 3",
],
)
self.assertEqual(saved["mode_profile"]["command_flow"]["execution_steps"], [])
self.assertEqual(executed["status"], "succeeded")
self.assertEqual(executed["utterance"], "办公模式")
self.assertEqual(executed["provider_used"], "saved_mode_command_flow")
self.assertEqual(executed["orchestration_route"], "saved_mode_command_flow")
self.assertEqual(
[item["slot"] for item in executed["infrared_actions"]],
[1, 3],
)
self.assertTrue(
all(
item["parse_source"] == "saved_mode_command_flow"
for item in executed["infrared_actions"]
)
)
def test_saved_mode_accepts_prefixed_self_extension_flow_key(self) -> None:
with tempfile.TemporaryDirectory() as directory:
extension = SelfExtensionService(
SimulationGateway(),
catalog_db_path=Path(directory, "prefixed-mode-ir.sqlite3"),
)
extension.sync_peripheral_catalog(
[
{
"id": "ir-slot:tony-light-001:1",
"status": "learned",
"enabled": True,
"device_id": "tony-light-001",
"device_name": "闪灯玩具",
"card_uid": "2CC8B103",
"action": "打开",
"slot": 1,
}
]
)
handler = object.__new__(SeatDemoHandler)
handler.extension = extension
handler.nfc = SimpleNamespace(
snapshot=lambda: {
"card": {"action": "present", "uid": "2CC8B103"}
}
)
profile = SimpleNamespace(
command_flow={
"function_configs": {
"local_self_extension": {
"spec": {
"domain": "self_extension",
"operations": [
{
"capability": "infrared.emit",
"target": {"device": "tony-light-001"},
"parameters": {"action": "打开", "slot": 1},
}
],
}
}
}
}
)
entries = handler._saved_mode_infrared_entries(profile)
self.assertEqual([item["capability"]["slot"] for item in entries], [1])
self.assertEqual(
entries[0]["capability"]["parse_source"],
"saved_mode_command_flow",
)
def test_head_unit_display_sync_tracks_tasks_and_usb_catalog(self) -> None:
task_request = urllib.request.Request(
f"{self.base_url}/api/requests",
@@ -344,14 +585,14 @@ operations:
def test_static_console_and_runtime_endpoint(self) -> None:
with self.opener.open(f"{self.base_url}/", timeout=3) as response:
html = response.read().decode("utf-8")
with self.opener.open(f"{self.base_url}/app.js?v=dual-seat-107", timeout=3) as response:
with self.opener.open(f"{self.base_url}/app.js?v=dual-seat-109", timeout=3) as response:
app_script = response.read().decode("utf-8")
with self.opener.open(
f"{self.base_url}/gesture-control.js?v=android-pad-9", timeout=3
) as response:
gesture_script = response.read().decode("utf-8")
with self.opener.open(
f"{self.base_url}/console-v2.js?v=delivery-state-49", timeout=3
f"{self.base_url}/console-v2.js?v=delivery-state-50", timeout=3
) as response:
console_script = response.read().decode("utf-8")
with self.opener.open(
@@ -394,11 +635,11 @@ operations:
self.assertIn("CMD 0x431 · FB 0x41A · CH1", html)
self.assertIn("CMD 0x431 · FB 0x388 · CH1", html)
self.assertNotIn("透明车体 · 实体设备", html)
self.assertIn('src="/app.js?v=dual-seat-107"', html)
self.assertIn('src="/app.js?v=dual-seat-109"', html)
self.assertIn('href="/styles.css?v=dual-seat-16"', html)
self.assertIn('src="/gesture-control.js?v=android-pad-9"', html)
self.assertNotIn('class="vehicle-system-hud"', html)
self.assertIn('src="/console-v2.js?v=delivery-state-49"', html)
self.assertIn('src="/console-v2.js?v=delivery-state-50"', html)
self.assertIn('href="/console-v2.css?v=delivery-state-43"', html)
self.assertIn('<body class="console-v2">', html)
self.assertIn('id="v2BootScreen"', html)
@@ -641,7 +882,8 @@ operations:
)
self.assertIn("const stageHasLayout = () =>", app_script)
self.assertIn("window.requestAnimationFrame(initSeatTwin)", app_script)
self.assertIn('client_type: "web_console"', app_script)
self.assertIn('clientType = "web_console"', app_script)
self.assertIn("client_type: clientType", app_script)
self.assertIn("function executionVerificationStatus(result = {})", app_script)
self.assertIn('"submitted_unverified"', app_script)
self.assertIn("指令已提交，等待状态确认", app_script)
@@ -649,8 +891,8 @@ operations:
self.assertIn("CAN 指令已提交，等待车辆状态确认", console_script)
self.assertIn('canvas.addEventListener("webglcontextlost"', app_script)
self.assertIn('"cognitive-agent:module-change"', app_script)
self.assertIn('const CACHE_VERSION = "cognitive-agent-pad-v218"', service_worker)
self.assertIn('"/app.js?v=dual-seat-107"', service_worker)
self.assertIn('const CACHE_VERSION = "cognitive-agent-pad-v220"', service_worker)
self.assertIn('"/app.js?v=dual-seat-109"', 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)
@@ -890,7 +1132,7 @@ operations:
) as response:
extension_script = response.read().decode("utf-8")
with self.opener.open(
f"{self.base_url}/app.js?v=dual-seat-107", timeout=3
f"{self.base_url}/app.js?v=dual-seat-109", timeout=3
) as response:
app_script = response.read().decode("utf-8")
tests/integration/test_multidomain_service.py
+176 -1
View File
@@ -621,7 +621,7 @@ operations:
if item["id"] == "office"
)
self.assertEqual(persisted["name"], "专注办公模式")
with self.assertRaisesRegex(ValueError, "重新点击执行"):
with self.assertRaisesRegex(ValueError, "生成指令"):
self.service.save_mode_profile_card(
"office",
"专注办公模式",
@@ -630,6 +630,181 @@ operations:
"denza_n7",
)
def test_saved_mode_card_executes_without_reinterpreting_instruction(self) -> None:
mode_root = Path(self.temporary_directory.name) / "direct-mode-presets"
user_root = Path(self.temporary_directory.name) / "direct-mode-user"
preset = mode_root / "vehicles" / "denza_n7" / "office.yaml"
preset.parent.mkdir(parents=True)
preset.write_text(
"""
id: office
name: 办公模式
icon: briefcase
scope:
vehicle_ids: [denza_n7]
operations:
- domain: climate
capability: climate.set_temperature
target_scope: driver
parameters:
temperature_c: 24
""".strip()
+ "\n",
encoding="utf-8",
)
self.service.mode_profiles = ModeProfileRegistry(
mode_root, user_root=user_root
)
instruction = "空调温度设置为25度，风速设置为1档"
generated = self.service.preview_native_can(
instruction,
provider="local",
conversation_id="conv-generate-office-card",
execution_mode="dry_run",
vehicle_id="denza_n7",
)
saved = self.service.save_mode_profile_card(
"office",
"办公模式",
instruction,
generated["request_id"],
"denza_n7",
)
with patch.object(
self.service,
"_interpret",
side_effect=AssertionError("saved mode must not reinterpret natural language"),
):
executed = self.service.execute_mode_profile_card(
"office",
conversation_id="conv-execute-office-card",
execution_mode="dry_run",
vehicle_id="denza_n7",
)
operations = [
operation
for config in executed.get("function_configs", {}).values()
for operation in config.get("spec", {}).get("operations", [])
]
self.assertEqual(executed["status"], "succeeded")
self.assertEqual(executed["utterance"], "办公模式")
self.assertEqual(executed["provider_used"], "saved_mode_command_flow")
self.assertEqual(executed["orchestration_route"], "saved_mode_command_flow")
self.assertEqual(
[
(
operation["capability"],
operation["target"]["zone"],
operation["parameters"],
operation["parse_source"],
)
for operation in operations
],
[
(
"climate.set_temperature",
"driver",
{"temperature_c": 25},
"saved_mode_command_flow",
),
(
"climate.set_fan_level",
"vehicle",
{"level": 1},
"saved_mode_command_flow",
),
],
)
self.assertEqual(
saved["mode_profile"]["command_flow"]["execution_steps"], []
)
def test_saved_mode_card_replays_front_right_dynamic_target_as_passenger(
self,
) -> None:
mode_root = Path(self.temporary_directory.name) / "dynamic-mode-presets"
user_root = Path(self.temporary_directory.name) / "dynamic-mode-user"
self.service.mode_profiles = ModeProfileRegistry(
mode_root,
user_root=user_root,
)
self.service.mode_profiles.save_mode(
"小憩模式",
[
{
"domain": "seat",
"capability": "seat.start_dynamic_mode",
"target_scope": "front_right",
"parameters": {"mode_id": 2},
},
{
"domain": "climate",
"capability": "climate.set_power",
"target_scope": "vehicle",
"parameters": {"state": "on"},
},
{
"domain": "roof",
"capability": "roof.set_shade_motion",
"target_scope": "vehicle",
"parameters": {"motion": "open"},
},
],
mode_id="nap",
vehicle_id="denza_n7",
instruction="打开摇篮模式，开启空调",
command_flow={
"schema": "cognitive-agent-mode-command-flow/v1",
"source_request_id": "req-saved-dynamic-mode",
"plans": {
"local_seat": {
"spec": {
"commands": [
{
"message": "Seat_Dynamic_Mode_Virtual",
"signals": {"Dynamic_Mode_ID": 2},
}
]
}
}
},
},
)
with patch.object(
self.service,
"_interpret",
side_effect=AssertionError("saved mode must not reinterpret natural language"),
):
executed = self.service.execute_mode_profile_card(
"nap",
conversation_id="conv-execute-saved-dynamic-card",
execution_mode="dry_run",
vehicle_id="denza_n7",
)
self.assertEqual(executed["status"], "succeeded")
self.assertEqual(executed["utterance"], "小憩模式")
self.assertEqual(executed["provider_used"], "saved_mode_command_flow")
self.assertFalse(
[item for item in executed.get("validation", []) if item["severity"] == "error"]
)
dynamic_request = next(
item
for item in executed["model_requests"]
if item["capability"] == "seat.start_dynamic_mode"
)
self.assertEqual(dynamic_request["target_scope"], "passenger")
self.assertTrue(executed["command_plans"])
replayed_capabilities = {
operation["capability"]
for config in executed["function_configs"].values()
for operation in config["spec"]["operations"]
}
self.assertIn("roof.set_shade_motion", replayed_capabilities)
def test_build_named_mode_executes_and_persists_in_one_turn(self) -> None:
mode_root = Path(self.temporary_directory.name) / "modes"
self.service.mode_profiles = ModeProfileRegistry(mode_root)
tests/unit/test_nfc_frontend_contract.py
+1 -1
View File
@@ -139,4 +139,4 @@ def test_main_parsed_control_panel_renders_validated_infrared_operations() -> No
assert 'frameLabel: "GSCU1B TCP"' in console
assert "已解析控制指令" in console
assert 'li[data-can-channel="IR"]' in styles
assert "console-v2.js?v=delivery-state-49" in html
assert "console-v2.js?v=delivery-state-50" in html
tests/unit/test_requested_frontend_enhancements.py
+13 -1
View File
@@ -437,7 +437,19 @@ def test_mode_editor_renders_readable_commands_with_expandable_mqtt_sequences()
assert 'source: "published"' in app
assert 'source: flow?.control_transport === "wireless_mqtt" ? "projected" : "can"' in app
assert "function_configs: functionConfigs" in app
assert "execution_steps: executionSteps" in app
assert "execution_steps: []" in app
assert "requestNaturalLanguageControl(instruction" in app
assert 'executionMode: "dry_run"' in app
assert 'clientType: "mode_editor_compile"' in app
assert "previewOnly: true" in app
assert "function executeSavedModeProfile(mode)" in app
assert "/api/mode-profiles/${encodeURIComponent(mode.id)}/execute" in app
assert '$("#utterance").value = mode.name || mode.id;' in app
assert 'const savedModeExecution = result.orchestration_route === "saved_mode_command_flow";' in app
assert "`${result.mode_profile.name}执行完成`" in app
assert "生成指令" in html
assert ">执行</span></button>" not in html
assert ".mode-editor-command-card" in styles
assert ".mode-editor-command-meta" in styles
tests/unit/test_self_extension.py
+10 -7
View File
@@ -970,7 +970,7 @@ class SelfExtensionTests(unittest.TestCase):
["failed"],
)
def test_submitted_emit_without_ack_continues_complete_infrared_plan(self) -> None:
def test_power_on_then_white_waits_800ms_and_continues_without_ack(self) -> None:
service = SelfExtensionService(SimulationGateway())
catalog = [
{
@@ -984,7 +984,7 @@ class SelfExtensionTests(unittest.TestCase):
}
for capability_id, action, slot in (
("light-on", "打开", 1),
("light-green", "切换到绿色", 4),
("light-white", "切换到白色", 6),
)
]
@@ -1009,16 +1009,18 @@ class SelfExtensionTests(unittest.TestCase):
"execute_gateway",
side_effect=submit_without_ack,
) as execute_gateway,
patch("cognitive_agent.application.self_extension.time.sleep"),
patch(
"cognitive_agent.application.self_extension.time.sleep"
) as sleep,
):
result = service.execute_peripheral_plan_request(
"开启闪光灯，颜色设置为绿色",
"启动闪烁灯，颜色设置为白色",
{
"provider_used": "infrared_local_rules",
"summary": "打开闪光灯并切换为绿色",
"summary": "启动闪烁灯并切换为白色",
"steps": [
{"kind": "peripheral", "capability_id": "light-on"},
{"kind": "peripheral", "capability_id": "light-green"},
{"kind": "peripheral", "capability_id": "light-white"},
],
},
catalog,
@@ -1027,11 +1029,12 @@ class SelfExtensionTests(unittest.TestCase):
)
self.assertEqual(execute_gateway.call_count, 2)
sleep.assert_called_once_with(0.8)
self.assertEqual(result["status"], "succeeded")
self.assertEqual(result["sequence"]["emit_count"], 2)
self.assertEqual(
[item["telemetry"]["slot"] for item in result["operation_results"]],
[1, 4],
[1, 6],
)
self.assertEqual(
[
tests/unit/test_two_stage_can_parsing.py
+1
View File
@@ -187,6 +187,7 @@ def test_frontend_two_stage_contract_and_source_badges() -> None:
assert "本地规则命令已执行；云端补充失败" in app
assert 'class="v2-can-command-source"' in console
assert 'item.parseSource === "direct_control" ? "设备控件"' in console
assert 'item.parseSource === "saved_mode_command_flow" ? "模式卡片"' in console
assert 'item.parseSource === "local_rule" ? "本地规则" : "云端模型"' in console
assert 'item.deliveryLabel' in console
assert 'data-delivery-state="${escapeHtml(item.deliveryState)}"' in console