minghanchen
320 lines
13 KiB
Python
320 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from cognitive_agent.application.mqtt_ota_seat_executor import MqttOtaSeatExecutor
|
|
from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService
|
|
|
|
|
|
CATALOG_PATH = (
|
|
REPOSITORY_ROOT
|
|
/ "tests"
|
|
/ "fixtures"
|
|
/ "natural_can_command_combination_suite.json"
|
|
)
|
|
HARDWARE_DISABLED_ENV = {
|
|
"COGNITIVE_AGENT_CH1_HIL_ENABLED": "false",
|
|
"COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false",
|
|
"COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false",
|
|
}
|
|
|
|
|
|
class NaturalCanCommandCombinationTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.temporary_directory = tempfile.TemporaryDirectory()
|
|
temporary_root = Path(cls.temporary_directory.name)
|
|
memory_path = temporary_root / "driver-seat-memory.json"
|
|
memory_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"version": 1,
|
|
"positions": {
|
|
str(slot): {
|
|
"payload_15": "15 00 00 00 00 00 00 00",
|
|
"payload_16": "16 00 00 00 00 00 00 00",
|
|
}
|
|
for slot in (1, 2, 3)
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
cls.environment = patch.dict(
|
|
os.environ,
|
|
{
|
|
**HARDWARE_DISABLED_ENV,
|
|
"COGNITIVE_AGENT_DRIVER_SEAT_MEMORY_PATH": str(memory_path),
|
|
},
|
|
clear=False,
|
|
)
|
|
cls.environment.start()
|
|
cls.service = SeatDemoService(REPOSITORY_ROOT)
|
|
cls.service.artifact_root = temporary_root / "artifacts"
|
|
cls.catalog = json.loads(CATALOG_PATH.read_text(encoding="utf-8"))
|
|
|
|
@classmethod
|
|
def tearDownClass(cls) -> None:
|
|
cls.environment.stop()
|
|
cls.temporary_directory.cleanup()
|
|
|
|
@staticmethod
|
|
def _debug_commands(result: dict) -> list[str]:
|
|
commands: list[str] = []
|
|
for plan in result.get("command_plans", {}).values():
|
|
for command in plan.get("spec", {}).get("commands", []):
|
|
debug_command = command.get("signals", {}).get("debug_command")
|
|
if isinstance(debug_command, str):
|
|
commands.append(debug_command)
|
|
return commands
|
|
|
|
@staticmethod
|
|
def _can0_operations(result: dict, expected: list[dict]) -> Counter[str]:
|
|
capabilities = {item["capability"] for item in expected}
|
|
operations: list[dict] = []
|
|
for config in result.get("function_configs", {}).values():
|
|
for operation in config.get("spec", {}).get("operations", []):
|
|
if operation.get("capability") not in capabilities:
|
|
continue
|
|
target = operation.get("target", {})
|
|
operations.append(
|
|
{
|
|
"target": target.get("instance", target.get("zone")),
|
|
"capability": operation["capability"],
|
|
"parameters": operation.get("parameters", {}),
|
|
}
|
|
)
|
|
return Counter(
|
|
json.dumps(item, ensure_ascii=False, sort_keys=True)
|
|
for item in operations
|
|
)
|
|
|
|
@staticmethod
|
|
def _expected_can0_operations(expected: list[dict]) -> Counter[str]:
|
|
return Counter(
|
|
json.dumps(item, ensure_ascii=False, sort_keys=True)
|
|
for item in expected
|
|
)
|
|
|
|
@staticmethod
|
|
def _single_passenger_motion_operation(result: dict) -> dict:
|
|
operations = [
|
|
operation
|
|
for config in result.get("function_configs", {}).values()
|
|
for operation in config.get("spec", {}).get("operations", [])
|
|
if operation.get("capability")
|
|
in {
|
|
"seat.set_longitudinal_position",
|
|
"seat.set_cushion_tilt",
|
|
"seat.set_backrest_recline",
|
|
"seat.set_legrest_height",
|
|
}
|
|
]
|
|
if len(operations) != 1:
|
|
raise AssertionError(f"expected one passenger motion operation: {operations}")
|
|
return operations[0]
|
|
|
|
def test_catalog_metadata_and_all_expanded_cases(self) -> None:
|
|
cases = self.catalog["cases"]
|
|
self.assertEqual(
|
|
self.catalog["schema"],
|
|
"natural-can-command-combination-suite/v1",
|
|
)
|
|
self.assertEqual(self.catalog["summary"]["total_cases"], len(cases))
|
|
self.assertGreaterEqual(len(cases), 1000)
|
|
self.assertEqual(len({item["id"] for item in cases}), len(cases))
|
|
|
|
for index, case in enumerate(cases):
|
|
with self.subTest(case=case["id"], utterance=case["utterance"]):
|
|
result = self.service.process(
|
|
case["utterance"],
|
|
provider="local",
|
|
conversation_id=f"conv-natural-combination-{index:04d}",
|
|
execution_mode="dry_run",
|
|
)
|
|
self.assertEqual(result["status"], case["expected_status"])
|
|
self.assertEqual(
|
|
self._debug_commands(result),
|
|
case["expected_commands"],
|
|
)
|
|
expected_can0 = case["expected_can0_operations"]
|
|
if expected_can0:
|
|
self.assertEqual(
|
|
self._can0_operations(result, expected_can0),
|
|
self._expected_can0_operations(expected_can0),
|
|
)
|
|
|
|
def test_can0_native_motion_and_dynamic_modes_combine_with_can1(self) -> None:
|
|
cases = (
|
|
(
|
|
"把副驾座椅水平位置调到35%，空调温度设置为24度，打开氛围灯",
|
|
["ac temp driver 24", "light ambient on"],
|
|
[{
|
|
"target": "passenger",
|
|
"capability": "seat.set_longitudinal_position",
|
|
"parameters": {"position_pct": 35},
|
|
}],
|
|
),
|
|
(
|
|
"副驾坐盆向上调整10%，关闭后排车窗，空调风量2档",
|
|
["window rear close", "ac fan 2"],
|
|
[{
|
|
"target": "passenger",
|
|
"capability": "seat.set_cushion_tilt",
|
|
"parameters": {"position_pct": 40},
|
|
}],
|
|
),
|
|
(
|
|
"开启木马模式并打开氛围灯",
|
|
["light ambient on"],
|
|
[{
|
|
"target": "passenger",
|
|
"capability": "seat.start_dynamic_mode",
|
|
"parameters": {"mode_id": 1},
|
|
}],
|
|
),
|
|
(
|
|
"开启动态摇篮模式，空调24度，打开遮阳帘",
|
|
["ac temp driver 24", "shade open"],
|
|
[{
|
|
"target": "passenger",
|
|
"capability": "seat.start_dynamic_mode",
|
|
"parameters": {"mode_id": 2},
|
|
}],
|
|
),
|
|
(
|
|
"副驾座椅前后调到20%、坐盆调到30%、靠背调到40%、腿托调到50%，空调风量3档，关闭全部车窗",
|
|
["ac fan 3", "window all close"],
|
|
[
|
|
{
|
|
"target": "passenger",
|
|
"capability": capability,
|
|
"parameters": {"position_pct": position},
|
|
}
|
|
for capability, position in (
|
|
("seat.set_longitudinal_position", 20),
|
|
("seat.set_cushion_tilt", 30),
|
|
("seat.set_backrest_recline", 40),
|
|
("seat.set_legrest_height", 50),
|
|
)
|
|
],
|
|
),
|
|
(
|
|
"副驾靠背向后移动15%，主驾温度设为25度，遮阳帘打开",
|
|
["ac temp driver 25", "shade open"],
|
|
[{
|
|
"target": "passenger",
|
|
"capability": "seat.set_backrest_recline",
|
|
"parameters": {"position_pct": 35},
|
|
}],
|
|
),
|
|
(
|
|
"副驾靠背往后转动10%，空调温度设置为24度",
|
|
["ac temp driver 24"],
|
|
[{
|
|
"target": "passenger",
|
|
"capability": "seat.set_backrest_recline",
|
|
"parameters": {"position_pct": 30},
|
|
}],
|
|
),
|
|
(
|
|
"副驾腿托设置为40%，关闭副驾车窗",
|
|
["window front-right close"],
|
|
[{
|
|
"target": "passenger",
|
|
"capability": "seat.set_legrest_height",
|
|
"parameters": {"position_pct": 40},
|
|
}],
|
|
),
|
|
)
|
|
|
|
for index, (utterance, can1_commands, can0_operations) in enumerate(cases):
|
|
with self.subTest(utterance=utterance):
|
|
result = self.service.process(
|
|
utterance,
|
|
provider="local",
|
|
conversation_id=f"conv-can0-can1-targeted-{index}",
|
|
execution_mode="dry_run",
|
|
)
|
|
self.assertEqual(result["status"], "succeeded")
|
|
self.assertEqual(self._debug_commands(result), can1_commands)
|
|
self.assertEqual(
|
|
self._can0_operations(result, can0_operations),
|
|
self._expected_can0_operations(can0_operations),
|
|
)
|
|
|
|
def test_passenger_relative_and_absolute_directional_commands_stay_distinct(self) -> None:
|
|
cases = (
|
|
("副驾座椅往上移动10%", "basin", "decrease", "up", 10),
|
|
("副驾座椅往上移动20%", "basin", "decrease", "up", 20),
|
|
("副驾往下移动10%", "basin", "increase", "down", 10),
|
|
("副驾往下移动20%", "basin", "increase", "down", 20),
|
|
("副驾座椅往前移动10%", "horizontal", "decrease", "forward", 10),
|
|
("副驾座椅往前移动20%", "horizontal", "decrease", "forward", 20),
|
|
("副驾往后移动10%", "horizontal", "increase", "backward", 10),
|
|
("副驾往后移动20%", "horizontal", "increase", "backward", 20),
|
|
("副驾座椅靠背往前移动10%", "backrest", "decrease", "forward", 10),
|
|
("副驾座椅靠背往前移动30%", "backrest", "decrease", "forward", 30),
|
|
("副驾靠背往后移动10%", "backrest", "increase", "backward", 10),
|
|
("副驾靠背往后移动30%", "backrest", "increase", "backward", 30),
|
|
("副驾推托往上移动20%", "legrest", "increase", "up", 20),
|
|
("副驾推托往上移动40%", "legrest", "increase", "up", 40),
|
|
("副驾推托往下移动20%", "legrest", "decrease", "down", 20),
|
|
("副驾推托往下移动40%", "legrest", "decrease", "down", 40),
|
|
)
|
|
|
|
for index, (utterance, axis, direction, command_direction, amount) in enumerate(
|
|
cases
|
|
):
|
|
with self.subTest(kind="relative", utterance=utterance):
|
|
result = self.service.process(
|
|
utterance,
|
|
provider="local",
|
|
conversation_id=f"conv-passenger-relative-{index}",
|
|
execution_mode="dry_run",
|
|
)
|
|
self.assertEqual(result["status"], "succeeded")
|
|
operation = self._single_passenger_motion_operation(result)
|
|
motion = operation["motion"]
|
|
self.assertEqual(motion["mode"], "relative")
|
|
self.assertEqual(motion["direction"], direction)
|
|
self.assertEqual(
|
|
motion.get(
|
|
"requested_delta_position_pct",
|
|
motion["delta_position_pct"],
|
|
),
|
|
amount,
|
|
)
|
|
self.assertTrue(
|
|
MqttOtaSeatExecutor._operation_debug_command(operation).startswith(
|
|
f"seat passenger {axis} relative {command_direction} "
|
|
)
|
|
)
|
|
|
|
absolute_utterance = utterance.replace("移动", "移动到")
|
|
with self.subTest(kind="absolute", utterance=absolute_utterance):
|
|
result = self.service.process(
|
|
absolute_utterance,
|
|
provider="local",
|
|
conversation_id=f"conv-passenger-absolute-{index}",
|
|
execution_mode="dry_run",
|
|
)
|
|
self.assertEqual(result["status"], "succeeded")
|
|
operation = self._single_passenger_motion_operation(result)
|
|
self.assertNotIn("motion", operation)
|
|
self.assertEqual(operation["parameters"]["position_pct"], amount)
|
|
self.assertEqual(
|
|
MqttOtaSeatExecutor._operation_debug_command(operation),
|
|
f"seat passenger {axis} position {amount}",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|