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()