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

feat: improve mode and peripheral orchestration

Browse Source
This commit is contained in:
Miayui
2026-09-21 00:13:47 +08:00
parent 89707a42e2
commit 448c8b31a0
39 changed files with 3068 additions and 277 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_mode_scene_ingress.py
+312
View File
@@ -205,6 +205,55 @@ def cinema_scene_card() -> dict[str, object]:
}
def cinema_peripheral_catalog() -> list[dict[str, object]]:
def capability(
identity: str,
device_id: str,
device_name: str,
action: str,
slot: int,
) -> dict[str, object]:
return {
"id": identity,
"status": "learned",
"enabled": True,
"device_id": device_id,
"device_name": device_name,
"device_aliases": [device_name],
"object_names": [device_name],
"card_uid": "2CC8B103" if "flash" in device_id else "1A3AB103",
"action": action,
"aliases": [action],
"action_names": [action],
"risk": "low",
"slot": slot,
}
return [
capability("flash-on", "flash-light", "闪烁灯", "打开", 1),
capability("flash-blue", "flash-light", "闪烁灯", "切换为蓝色", 3),
capability("flash-red", "flash-light", "闪烁灯", "切换为红色", 4),
capability("strip-on", "light-strip", "灯带", "打开", 21),
capability("strip-blue", "light-strip", "灯带", "切换为蓝色", 23),
capability("strip-red", "light-strip", "灯带", "切换为红色", 24),
]
def cinema_video_launch_action() -> dict[str, object]:
return {
"type": "android_app_launch",
"at_ms": 3000,
"domain": "android",
"capability": "android.launch_app",
"target_scope": "headunit",
"operation": "launch",
"parameters": {
"activity_name": "com.byd.videoplay.activity.WelComeActivity",
"package_name": "com.byd.videoplay",
},
}
def wait_until(predicate, timeout: float = 1.5) -> bool:
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
@@ -337,6 +386,269 @@ def test_import_upserts_yaml_refreshes_catalog_and_does_not_echo(
assert "upstream_origin: user" in yaml_text
def test_later_n7_cinema_mode_inherits_base_and_adds_external_light_colors(
tmp_path: Path,
) -> None:
service = configured_service(tmp_path)
service.set_mode_scene_peripheral_catalog_provider(cinema_peripheral_catalog)
base = cinema_scene_card()
base_action_count = len(base["actions"])
base_result = service.import_n7_mode_scene(base)
assert "cinema_augmentation" not in base_result
variant = cinema_scene_card()
variant.update(
id="blue-cinema",
label="蓝色影院模式",
name="蓝色影院模式",
aliases=["蓝色观影模式"],
description="在车机影院模式基础上增加蓝色外设灯光。",
domains=["climate"],
operation_count=1,
actions=[
{
"type": "climate_temperature",
"at_ms": 0,
"domain": "climate",
"capability": "climate.set_temperature",
"target_scope": "vehicle",
"operation": "set",
"parameters": {"temperature_c": 23},
}
],
profile_revision="blue-cinema-revision",
)
imported = service.import_n7_mode_scene(variant)
augmentation = imported["cinema_augmentation"]
assert augmentation == {
"base_scene_id": "cinema",
"base_action_count": base_action_count,
"incoming_action_count": 1,
"peripheral_action_count": 4,
"merged_action_count": base_action_count + 4,
}
profile = service.get_mode_profile_card("blue-cinema", "denza_n7")
source_scene = profile.command_flow["source_scene"]
actions = source_scene["actions"]
assert source_scene["operation_count"] == base_action_count + 4
assert "self_extension" in source_scene["domains"]
temperatures = [
action["parameters"]["temperature_c"]
for action in actions
if action["capability"] == "climate.set_temperature"
]
assert temperatures == [23]
peripheral_actions = [
action
for action in actions
if action["capability"] == "infrared.emit"
]
assert [
(
action["target_scope"],
action["parameters"]["action"],
action["parameters"]["slot"],
)
for action in peripheral_actions
] == [
("flash-light", "打开", 1),
("flash-light", "切换为蓝色", 3),
("light-strip", "打开", 21),
("light-strip", "切换为蓝色", 23),
]
repeated = service.import_n7_mode_scene(variant)
assert (
repeated["cinema_augmentation"]["merged_action_count"]
== base_action_count + 4
)
reloaded = service.get_mode_profile_card("blue-cinema", "denza_n7")
assert (
len(reloaded.command_flow["source_scene"]["actions"])
== base_action_count + 4
)
def test_n7_cinema_variant_is_not_augmented_before_base_arrives(
tmp_path: Path,
) -> None:
service = configured_service(tmp_path)
service.set_mode_scene_peripheral_catalog_provider(cinema_peripheral_catalog)
variant = cinema_scene_card()
variant.update(
id="family-movie",
name="家庭影院模式",
label="家庭影院模式",
profile_revision="family-movie-revision",
)
imported = service.import_n7_mode_scene(variant)
assert "cinema_augmentation" not in imported
profile = service.get_mode_profile_card("family-movie", "denza_n7")
assert not any(
operation.capability == "infrared.emit"
for operation in profile.operations
)
def test_created_romantic_cinema_persists_and_transfers_n7_video_launch(
tmp_path: Path,
) -> None:
service = configured_service(tmp_path)
base = cinema_scene_card()
base["actions"] = [*base["actions"], cinema_video_launch_action()]
base["operation_count"] = len(base["actions"])
base["domains"] = [*base["domains"], "android"]
service.import_n7_mode_scene(base)
request_id = "req-romantic-cinema-video"
instruction = "进入浪漫观影模式"
result_dir = service.artifact_root / request_id
result_dir.mkdir(parents=True)
(result_dir / "result.json").write_text(
json.dumps(
{
"request_id": request_id,
"status": "succeeded",
"execution_mode": "dry_run",
"control_transport": "wireless_mqtt",
"utterance": instruction,
"vehicle": {"id": "denza_n7"},
"function_configs": {
"climate": {
"spec": {
"domain": "climate",
"operations": [
{
"id": "vehicle-climate-power",
"capability": "climate.set_power",
"target": {"zone": "vehicle"},
"parameters": {"state": "on"},
}
],
}
}
},
"command_plans": {
"climate": {
"spec": {
"commands": [
{
"operation_ref": "vehicle-climate-power",
"message": "climate.set_power",
}
]
}
}
},
},
ensure_ascii=False,
),
encoding="utf-8",
)
saved = service.create_mode_profile_card(
"浪漫观影模式",
instruction,
request_id,
"denza_n7",
)
profile = service.get_mode_profile_card(
str(saved["mode_profile"]["id"]),
"denza_n7",
)
assert [
operation.capability for operation in profile.operations
] == ["climate.set_power", "android.launch_app"]
assert profile.command_flow["n7_transfer_actions"] == [
cinema_video_launch_action()
]
local_requests, transfer_actions = service._mode_requests_and_n7_transfers(
profile
)
assert [request["capability"] for request in local_requests] == [
"climate.set_power"
]
assert transfer_actions == [cinema_video_launch_action()]
exported = next(
scene
for scene in service.mode_scene_catalog("denza_n7")["scenes"]
if scene["id"] == profile.id
)
video = next(
action
for action in exported["actions"]
if action["capability"] == "android.launch_app"
)
assert video == cinema_video_launch_action()
def test_existing_local_cinema_mode_publishes_video_action_only_in_hardware(
tmp_path: Path,
) -> None:
service = configured_service(tmp_path)
base = cinema_scene_card()
base["actions"] = [*base["actions"], cinema_video_launch_action()]
base["operation_count"] = len(base["actions"])
base["domains"] = [*base["domains"], "android"]
service.import_n7_mode_scene(base)
profile = service.mode_profiles.save_mode(
"浪漫观影模式",
[
{
"domain": "android",
"capability": "android.launch_app",
"target_scope": "headunit",
"operation": "launch",
"parameters": deepcopy(cinema_video_launch_action()["parameters"]),
}
],
vehicle_id="denza_n7",
instruction="进入浪漫观影模式",
command_flow={
"schema": "cognitive-agent-mode-command-flow/v1",
"plans": {"saved": {"spec": {"commands": [{"message": "saved"}]}}},
},
)
publisher = MagicMock(
return_value={
"topic": "transto_n7/ctl",
"qos": 1,
"retain": False,
"delivery_status": "published",
"message_id": 19,
}
)
service._vehicle_mqtt_transport.publish_n7_control_transfer = publisher
preview = service.execute_mode_profile_card(
profile.id,
execution_mode="dry_run",
control_transport="wireless_mqtt",
)
publisher.assert_not_called()
assert preview["n7_transfer_candidates"] == [cinema_video_launch_action()]
executed = service.execute_mode_profile_card(
profile.id,
execution_mode="hardware",
control_transport="wireless_mqtt",
)
assert executed["n7_control_transfer"]["state"] == "published"
publisher.assert_called_once()
payload = publisher.call_args.args[0]
assert payload["scene_id"] == profile.id
assert payload["scene_name"] == "浪漫观影模式"
assert payload["operation_count"] == 1
assert payload["actions"] == [cinema_video_launch_action()]
def test_unknown_n7_action_is_forwarded_once_only_during_hardware_execution(
tmp_path: Path,
) -> None: