vee
662 lines
30 KiB
Python
662 lines
30 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from cognitive_agent.application.native_can_intent import merge_native_can_intent
|
|
from cognitive_agent.application.native_can_runtime import NativeCanRuntime
|
|
from cognitive_agent.infrastructure.speech.capability import CapabilityResolver
|
|
from cognitive_agent.infrastructure.speech.understanding import (
|
|
UnderstandingTranscriptProcessor,
|
|
)
|
|
from cognitive_agent.infrastructure.speech.vocabulary import DeviceTerm
|
|
|
|
REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def _valid_value(spec: dict[str, Any]) -> Any:
|
|
allowed = spec.get("allowed_values")
|
|
if allowed:
|
|
return allowed[0]
|
|
value_type = str(spec.get("type", "string"))
|
|
if value_type in {"integer", "number"}:
|
|
minimum = spec.get("minimum", 0)
|
|
return int(minimum) if value_type == "integer" else float(minimum)
|
|
if value_type == "boolean":
|
|
return True
|
|
if value_type == "object":
|
|
return {}
|
|
if value_type == "array":
|
|
return []
|
|
if str(spec.get("name")) == "color_hex":
|
|
return "#FF0000"
|
|
if str(spec.get("name")) == "preset_name":
|
|
return "回归测试姿态"
|
|
return "test"
|
|
|
|
|
|
def _sample_parameters(definition: Any) -> dict[str, Any]:
|
|
specs = list(definition.parameters)
|
|
selected = [item for item in specs if item.get("required")]
|
|
if not selected and specs:
|
|
selected = [specs[0]]
|
|
return {str(item["name"]): _valid_value(item) for item in selected}
|
|
|
|
|
|
class SpeechCapabilityAlignmentTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.processor = UnderstandingTranscriptProcessor(REPOSITORY_ROOT)
|
|
cls.resolver = CapabilityResolver(REPOSITORY_ROOT)
|
|
cls.native_runtime = NativeCanRuntime(REPOSITORY_ROOT)
|
|
|
|
def test_eight_reported_cases_preserve_structure_and_resolution(self) -> None:
|
|
cases = (
|
|
{
|
|
"text": "把遮阳帘打开",
|
|
"status": "recognized_executable",
|
|
"clean": "打开遮阳帘。",
|
|
"commands": (
|
|
(
|
|
"roof.set_shade_motion",
|
|
"vehicle",
|
|
"roof.shade_motion",
|
|
"open",
|
|
{"motion": "open"},
|
|
),
|
|
),
|
|
},
|
|
{
|
|
"text": "打开前大灯",
|
|
"status": "recognized_unsupported",
|
|
"clean": "打开前大灯。",
|
|
"commands": ((None, "vehicle", "front_light.power", "on", {"state": "on"}),),
|
|
},
|
|
{
|
|
"text": "打开主驾座椅通风",
|
|
"status": "recognized_executable",
|
|
"clean": "打开主驾座椅通风。",
|
|
"commands": (
|
|
(
|
|
"seat.set_driver_ventilation_native",
|
|
"driver",
|
|
"seat.ventilation",
|
|
"on",
|
|
{"state": "on", "level": 2},
|
|
),
|
|
),
|
|
},
|
|
{
|
|
"text": "把主驾座椅调整为零重力模式",
|
|
"status": "recognized_unsupported",
|
|
"clean": "将主驾座椅调整为零重力模式。",
|
|
"commands": (
|
|
(None, "driver", "seat.mode", "apply", {"mode_profile_id": "zero_gravity"}),
|
|
),
|
|
},
|
|
{
|
|
"text": "打开一半车窗",
|
|
"status": "recognized_executable",
|
|
"clean": "将车窗打开到50%。",
|
|
"commands": (
|
|
(
|
|
"window.set_position",
|
|
"front_left",
|
|
"window.position",
|
|
"set",
|
|
{"position_pct": 50},
|
|
),
|
|
),
|
|
},
|
|
{
|
|
"text": "打开氛围灯并调整为红色",
|
|
"status": "incomplete",
|
|
"clean": "打开氛围灯，然后将氛围灯颜色调整为红色。",
|
|
"commands": (
|
|
("ambient_light.set_power", "vehicle", "ambient.power", "on", {"state": "on"}),
|
|
(
|
|
"ambient_light.set_illumination",
|
|
"vehicle",
|
|
"ambient.color",
|
|
"set",
|
|
{"color_hex": "#FF0000"},
|
|
),
|
|
),
|
|
},
|
|
{
|
|
"text": "打开主驾座椅加热",
|
|
"status": "recognized_executable",
|
|
"clean": "打开主驾座椅加热。",
|
|
"commands": (
|
|
(
|
|
"seat.set_driver_heating_native",
|
|
"driver",
|
|
"seat.heating",
|
|
"on",
|
|
{"state": "on", "level": 2},
|
|
),
|
|
),
|
|
},
|
|
{
|
|
"text": "打开副驾座椅通风",
|
|
"status": "recognized_executable",
|
|
"clean": "打开副驾座椅通风。",
|
|
"commands": (
|
|
(
|
|
"seat.set_ventilation",
|
|
"passenger",
|
|
"seat.ventilation",
|
|
"on",
|
|
{"state": "on", "level": 1},
|
|
),
|
|
),
|
|
},
|
|
)
|
|
for case in cases:
|
|
with self.subTest(text=case["text"]):
|
|
result = self.processor.process(case["text"])
|
|
self.assertEqual(result.raw_text, case["text"])
|
|
self.assertEqual(result.clean_text, case["clean"])
|
|
self.assertEqual(result.semantic_status, case["status"])
|
|
self.assertEqual(len(result.commands), len(case["commands"]))
|
|
for command, expected in zip(result.commands, case["commands"], strict=True):
|
|
capability, target, property_name, operation, parameters = expected
|
|
self.assertEqual(command.capability, capability)
|
|
self.assertEqual(command.target_scope, target)
|
|
self.assertEqual(command.property_name, property_name)
|
|
self.assertEqual(command.operation, operation)
|
|
self.assertEqual(command.parameters, parameters)
|
|
if command.resolution_status == "executable":
|
|
self.assertIsNotNone(command.intent_ir_action)
|
|
self.assertIsNotNone(command.executor_payload)
|
|
self.assertEqual(command.intent_ir_action["target"]["zone"], target)
|
|
self.assertEqual(command.executor_payload["target"], target)
|
|
self.assertEqual(command.executor_payload["parameters"], parameters)
|
|
else:
|
|
self.assertIsNone(command.executor_payload)
|
|
|
|
def test_semantic_trace_covers_every_required_boundary(self) -> None:
|
|
result = self.processor.process("打开主驾座椅通风")
|
|
self.assertEqual(
|
|
[item["stage"] for item in result.semantic_trace],
|
|
[
|
|
"raw_text",
|
|
"normalized_text",
|
|
"clause_segmentation",
|
|
"semantic_commands",
|
|
"intent_ir",
|
|
"capability_resolution",
|
|
"executor_payload",
|
|
"clean_text",
|
|
],
|
|
)
|
|
|
|
def test_explicit_control_slots_cannot_disappear_from_clean_text(self) -> None:
|
|
cases = (
|
|
("打开主驾座椅加热", ("主驾", "座椅加热")),
|
|
("打开副驾座椅通风", ("副驾", "座椅通风")),
|
|
("打开右后车窗", ("右后", "车窗")),
|
|
("打开一半车窗", ("50%", "车窗")),
|
|
("把氛围灯调整为红色", ("氛围灯", "红色")),
|
|
("把主驾座椅调整为零重力模式", ("主驾", "零重力模式")),
|
|
)
|
|
for raw, preserved in cases:
|
|
with self.subTest(raw=raw):
|
|
result = self.processor.process(raw)
|
|
for token in preserved:
|
|
self.assertIn(token, result.clean_text)
|
|
|
|
def test_compound_shared_target_is_preserved_and_gated_atomically(self) -> None:
|
|
result = self.processor.process("打开氛围灯并调整为红色")
|
|
self.assertEqual(
|
|
[command.property_name for command in result.commands],
|
|
["ambient.power", "ambient.color"],
|
|
)
|
|
self.assertEqual(result.commands[1].value, "#FF0000")
|
|
self.assertEqual(result.commands[1].resolution_status, "incomplete")
|
|
self.assertIn("brightness_pct", "".join(result.commands[1].validation_issues))
|
|
self.assertFalse(result.execution_eligible)
|
|
|
|
complete = self.processor.process("把氛围灯调整为红色并将亮度设置为50%")
|
|
self.assertEqual(len(complete.commands), 1)
|
|
self.assertEqual(
|
|
complete.commands[0].parameters,
|
|
{"color_hex": "#FF0000", "brightness_pct": 50},
|
|
)
|
|
self.assertEqual(complete.semantic_status, "recognized_executable")
|
|
self.assertIn("红色", complete.clean_text)
|
|
self.assertIn("50%", complete.clean_text)
|
|
|
|
def test_reported_compound_and_extreme_position_phrases_preserve_all_slots(self) -> None:
|
|
compound = self.processor.process(
|
|
"打开阅读灯，并且打开氛围灯，并把氛围灯变成红色"
|
|
)
|
|
self.assertEqual(
|
|
[command.property_name for command in compound.commands],
|
|
["reading_light.power", "ambient.power", "ambient.color"],
|
|
)
|
|
self.assertEqual(compound.commands[-1].parameters, {"color_hex": "#FF0000"})
|
|
self.assertEqual(compound.semantic_status, "incomplete")
|
|
self.assertFalse(compound.execution_eligible)
|
|
for token in ("阅读灯", "氛围灯", "红色"):
|
|
self.assertIn(token, compound.clean_text)
|
|
|
|
legrest = self.processor.process("把副驾腿托向上调整到最大位置")
|
|
self.assertEqual(legrest.semantic_status, "recognized_executable")
|
|
self.assertTrue(legrest.execution_eligible)
|
|
self.assertEqual(legrest.commands[0].target_scope, "passenger")
|
|
self.assertEqual(legrest.commands[0].capability, "seat.set_legrest_height")
|
|
self.assertEqual(legrest.commands[0].parameters, {"position_pct": 100})
|
|
for token in ("副驾", "腿托", "向上", "100%"):
|
|
self.assertIn(token, legrest.clean_text)
|
|
|
|
def test_directional_seat_endpoints_match_typed_axis_polarity(self) -> None:
|
|
cases = (
|
|
("把副驾滑轨向前调整到最大位置", "seat.longitudinal_position", 0),
|
|
("把副驾座椅向后移动至极限位置", "seat.longitudinal_position", 100),
|
|
("把副驾坐垫向上调整到最大位置", "seat.cushion_tilt", 0),
|
|
("把副驾座椅高度往下移动至极限位置", "seat.cushion_tilt", 100),
|
|
("把副驾靠背向前调整到最大位置", "seat.backrest_recline", 0),
|
|
("把副驾靠背向后移动至极限位置", "seat.backrest_recline", 100),
|
|
("把副驾腿托向上调整到最大位置", "seat.legrest_height", 100),
|
|
("把副驾脚托向下移动至极限位置", "seat.legrest_height", 0),
|
|
)
|
|
for text, property_name, position_pct in cases:
|
|
with self.subTest(text=text):
|
|
result = self.processor.process(text)
|
|
self.assertEqual(result.semantic_status, "recognized_executable")
|
|
self.assertTrue(result.execution_eligible)
|
|
self.assertEqual(len(result.commands), 1)
|
|
self.assertEqual(result.commands[0].property_name, property_name)
|
|
self.assertEqual(
|
|
result.commands[0].parameters,
|
|
{"position_pct": position_pct},
|
|
)
|
|
|
|
def test_latest_reported_compounds_preserve_every_mutation(self) -> None:
|
|
passenger = self.processor.process(
|
|
"打开副驾座椅通风和加热，，然后将副驾座椅靠背调整到最大位置"
|
|
)
|
|
self.assertEqual(
|
|
[command.property_name for command in passenger.commands],
|
|
["seat.ventilation", "seat.heating", "seat.backrest_recline"],
|
|
)
|
|
self.assertEqual(
|
|
[command.target_scope for command in passenger.commands],
|
|
["passenger", "passenger", "passenger"],
|
|
)
|
|
self.assertEqual(passenger.commands[-1].parameters, {"position_pct": 100})
|
|
self.assertTrue(passenger.execution_eligible)
|
|
for token in ("副驾座椅通风", "副驾座椅加热", "副驾座椅靠背", "100%"):
|
|
self.assertIn(token, passenger.clean_text)
|
|
|
|
driver = self.processor.process(
|
|
"打开主驾座椅通风和加热，并且把空调调整为24度"
|
|
)
|
|
self.assertEqual(
|
|
[command.property_name for command in driver.commands],
|
|
["seat.ventilation", "seat.heating", "hvac.temperature"],
|
|
)
|
|
self.assertEqual(
|
|
[command.target_scope for command in driver.commands[:2]],
|
|
["driver", "driver"],
|
|
)
|
|
self.assertEqual(driver.commands[-1].parameters, {"temperature_c": 24})
|
|
self.assertTrue(driver.execution_eligible)
|
|
for token in ("主驾座椅通风", "主驾座椅加热", "24度"):
|
|
self.assertIn(token, driver.clean_text)
|
|
|
|
def test_coordinated_seat_properties_cover_order_connectors_and_actions(self) -> None:
|
|
cases = (
|
|
("打开主驾座椅通风与加热", "driver", ("seat.ventilation", "seat.heating"), ("on", "on")),
|
|
("打开副驾座椅加热以及通风", "passenger", ("seat.heating", "seat.ventilation"), ("on", "on")),
|
|
("关闭主驾座椅通风、加热", "driver", ("seat.ventilation", "seat.heating"), ("off", "off")),
|
|
("打开主驾座椅通风和关闭加热", "driver", ("seat.ventilation", "seat.heating"), ("on", "off")),
|
|
)
|
|
for text, scope, properties, operations in cases:
|
|
with self.subTest(text=text):
|
|
result = self.processor.process(text)
|
|
self.assertEqual(
|
|
tuple(command.property_name for command in result.commands),
|
|
properties,
|
|
)
|
|
self.assertEqual(
|
|
tuple(command.operation for command in result.commands),
|
|
operations,
|
|
)
|
|
self.assertEqual(
|
|
{command.target_scope for command in result.commands},
|
|
{scope},
|
|
)
|
|
self.assertTrue(result.execution_eligible)
|
|
self.assertTrue(all(command.executor_payload for command in result.commands))
|
|
|
|
def test_coordinated_targets_keep_local_actions_scopes_and_properties(self) -> None:
|
|
cases = (
|
|
(
|
|
"关闭空调和打开氛围灯",
|
|
(("hvac.power", "off", "vehicle"), ("ambient.power", "on", "vehicle")),
|
|
),
|
|
(
|
|
"打开主驾座椅通风和副驾座椅加热",
|
|
(("seat.ventilation", "on", "driver"), ("seat.heating", "on", "passenger")),
|
|
),
|
|
(
|
|
"打开前排和后排空调",
|
|
(("hvac.power", "on", "front_row"), ("hvac.power", "on", "rear_row")),
|
|
),
|
|
(
|
|
"打开主驾和副驾车窗",
|
|
(("window.motion", "open", "front_left"), ("window.motion", "open", "front_right")),
|
|
),
|
|
)
|
|
for text, expected in cases:
|
|
with self.subTest(text=text):
|
|
result = self.processor.process(text)
|
|
self.assertEqual(
|
|
tuple(
|
|
(command.property_name, command.operation, command.target_scope)
|
|
for command in result.commands
|
|
),
|
|
expected,
|
|
)
|
|
self.assertTrue(result.execution_eligible)
|
|
for command in result.commands:
|
|
self.assertEqual(
|
|
command.executor_payload["target"], command.target_scope
|
|
)
|
|
|
|
def test_ratio_semantics_are_shared_across_continuous_capabilities(self) -> None:
|
|
cases = (
|
|
("打开四分之一车窗", "window.position", {"position_pct": 25}, "25%"),
|
|
("打开四分之三车窗", "window.position", {"position_pct": 75}, "75%"),
|
|
("打开百分之三十车窗", "window.position", {"position_pct": 30}, "30%"),
|
|
("打开1/4车窗", "window.position", {"position_pct": 25}, "25%"),
|
|
("把遮阳帘打开到四分之一", "roof.shade_position", {"position_pct": 25}, "25%"),
|
|
("把副驾腿托调整到四分之三", "seat.legrest_height", {"position_pct": 75}, "75%"),
|
|
("把副驾座椅向后调整50%", "seat.longitudinal_position", {"position_pct": 50}, "50%"),
|
|
)
|
|
for text, property_name, parameters, display in cases:
|
|
with self.subTest(text=text):
|
|
result = self.processor.process(text)
|
|
self.assertEqual(len(result.commands), 1)
|
|
self.assertEqual(result.commands[0].property_name, property_name)
|
|
self.assertEqual(result.commands[0].parameters, parameters)
|
|
self.assertIn(display, result.clean_text)
|
|
self.assertTrue(result.execution_eligible)
|
|
|
|
def test_invalid_unit_keeps_all_spoken_slots_and_fails_closed(self) -> None:
|
|
raw = "打开遮阳帘，然后把副驾座椅向后调整50度"
|
|
result = self.processor.process(raw)
|
|
self.assertEqual(result.semantic_status, "incomplete")
|
|
self.assertFalse(result.execution_eligible)
|
|
self.assertEqual(result.commands[1].target_scope, "passenger")
|
|
self.assertEqual(result.commands[1].value, 50)
|
|
self.assertEqual(result.commands[1].parameters, {})
|
|
self.assertIsNone(result.commands[1].executor_payload)
|
|
for token in ("遮阳帘", "副驾", "座椅", "向后", "50度"):
|
|
self.assertIn(token, result.clean_text)
|
|
self.assertNotIn("将副驾座椅向后调整。", result.clean_text)
|
|
|
|
def test_shared_value_and_coordination_grammar_also_feed_one_line_parser(self) -> None:
|
|
window = merge_native_can_intent({}, "打开四分之一车窗")
|
|
self.assertEqual(
|
|
[
|
|
(action["capability"], action["target"], action["parameters"])
|
|
for action in window["native_can_actions"]
|
|
],
|
|
[("window.set_position", "front_left", {"position_pct": 25})],
|
|
)
|
|
|
|
thermal = merge_native_can_intent({}, "打开主驾座椅通风和加热")
|
|
self.assertEqual(
|
|
[action["capability"] for action in thermal["native_can_actions"]],
|
|
[
|
|
"seat.set_driver_ventilation_native",
|
|
"seat.set_driver_heating_native",
|
|
],
|
|
)
|
|
self.assertEqual(
|
|
[action["parameters"] for action in thermal["native_can_actions"]],
|
|
[
|
|
{"state": "on", "level": 2},
|
|
{"state": "on", "level": 2},
|
|
],
|
|
)
|
|
|
|
def test_shared_target_inherits_resolved_driver_scope_and_level_semantics(self) -> None:
|
|
result = self.processor.process("打开座椅通风并调到三档")
|
|
self.assertEqual(
|
|
[command.capability for command in result.commands],
|
|
[
|
|
"seat.set_driver_ventilation_native",
|
|
"seat.set_driver_ventilation_native",
|
|
],
|
|
)
|
|
self.assertEqual(
|
|
[command.target_scope for command in result.commands],
|
|
["driver", "driver"],
|
|
)
|
|
self.assertIn("3档", result.clean_text)
|
|
self.assertEqual(result.semantic_status, "recognized_unsupported")
|
|
self.assertFalse(result.execution_eligible)
|
|
|
|
passenger = self.processor.process("打开副驾座椅通风并调到三档")
|
|
self.assertEqual(
|
|
[command.target_scope for command in passenger.commands],
|
|
["passenger", "passenger"],
|
|
)
|
|
self.assertTrue(passenger.execution_eligible)
|
|
|
|
graded = self.processor.process("打开主驾座椅加热并调到高档")
|
|
self.assertEqual(graded.commands[-1].parameters["level"], 1)
|
|
self.assertIn("高档", graded.clean_text)
|
|
|
|
def test_colloquial_seat_aliases_do_not_turn_pragmatic_one_into_level(self) -> None:
|
|
cases = (
|
|
(
|
|
"把驾驶座通风打开",
|
|
"driver",
|
|
"seat.set_driver_ventilation_native",
|
|
{"state": "on", "level": 2},
|
|
),
|
|
(
|
|
"把乘客座通风打开",
|
|
"passenger",
|
|
"seat.set_ventilation",
|
|
{"state": "on", "level": 1},
|
|
),
|
|
(
|
|
"给副驾驶开一下座椅通风",
|
|
"passenger",
|
|
"seat.set_ventilation",
|
|
{"state": "on", "level": 1},
|
|
),
|
|
)
|
|
for text, target, capability, parameters in cases:
|
|
with self.subTest(text=text):
|
|
result = self.processor.process(text)
|
|
self.assertTrue(result.execution_eligible)
|
|
self.assertEqual(len(result.commands), 1)
|
|
self.assertEqual(result.commands[0].target_scope, target)
|
|
self.assertEqual(result.commands[0].capability, capability)
|
|
self.assertEqual(result.commands[0].parameters, parameters)
|
|
self.assertIn(
|
|
"level",
|
|
result.commands[0].qualifiers.get("implicit_parameters", ()),
|
|
)
|
|
self.assertNotIn("设置为", result.clean_text)
|
|
|
|
explicit_level = self.processor.process("主驾座椅通风开到一档")
|
|
self.assertEqual(explicit_level.commands[0].parameters["level"], 1)
|
|
|
|
def test_discourse_connectors_preserve_actions_without_leaking_scope(self) -> None:
|
|
cases = (
|
|
"打开主驾座椅通风，然后打开副驾座椅通风，然后将空调设置为24度。",
|
|
"先打开主驾座椅通风，再打开副驾座椅通风，最后把空调调成二十四度",
|
|
)
|
|
expected = (
|
|
("seat.set_driver_ventilation_native", "driver", {"state": "on", "level": 2}),
|
|
("seat.set_ventilation", "passenger", {"state": "on", "level": 1}),
|
|
("climate.set_temperature", "driver", {"temperature_c": 24}),
|
|
)
|
|
for text in cases:
|
|
with self.subTest(text=text):
|
|
result = self.processor.process(text)
|
|
self.assertEqual(
|
|
tuple(
|
|
(command.capability, command.target_scope, command.parameters)
|
|
for command in result.commands
|
|
),
|
|
expected,
|
|
)
|
|
self.assertTrue(result.execution_eligible)
|
|
|
|
for text in (
|
|
"打开主驾座椅通风还有副驾座椅通风",
|
|
"打开主驾座椅通风之后打开副驾座椅通风",
|
|
"打开主驾通风，再给副驾也打开",
|
|
):
|
|
with self.subTest(text=text):
|
|
result = self.processor.process(text)
|
|
self.assertEqual(
|
|
[(command.capability, command.target_scope) for command in result.commands],
|
|
[
|
|
("seat.set_driver_ventilation_native", "driver"),
|
|
("seat.set_ventilation", "passenger"),
|
|
],
|
|
)
|
|
self.assertTrue(result.execution_eligible)
|
|
|
|
def test_elliptical_values_inherit_only_an_authoritative_property(self) -> None:
|
|
temperature = self.processor.process("空调二十四度")
|
|
self.assertEqual(len(temperature.commands), 1)
|
|
self.assertEqual(temperature.commands[0].capability, "climate.set_temperature")
|
|
self.assertEqual(temperature.commands[0].target_scope, "driver")
|
|
self.assertEqual(temperature.commands[0].parameters, {"temperature_c": 24})
|
|
self.assertTrue(temperature.execution_eligible)
|
|
|
|
continued = self.processor.process("打开空调，然后二十四度")
|
|
self.assertEqual(
|
|
[(command.capability, command.target_scope) for command in continued.commands],
|
|
[("climate.set_power", "vehicle"), ("climate.set_temperature", "driver")],
|
|
)
|
|
self.assertTrue(continued.execution_eligible)
|
|
|
|
ambient = self.processor.process("氛围灯打开，调成蓝色，亮度百分之五十")
|
|
self.assertEqual(
|
|
[command.capability for command in ambient.commands],
|
|
["ambient_light.set_power", "ambient_light.set_illumination"],
|
|
)
|
|
self.assertEqual(
|
|
ambient.commands[-1].parameters,
|
|
{"color_hex": "#0000FF", "brightness_pct": 50},
|
|
)
|
|
self.assertTrue(ambient.execution_eligible)
|
|
|
|
def test_non_control_continuation_never_replays_previous_action(self) -> None:
|
|
for text in ("打开空调，然后开玩笑", "打开空调，然后开会"):
|
|
with self.subTest(text=text):
|
|
result = self.processor.process(text)
|
|
self.assertEqual(
|
|
[command.capability for command in result.commands],
|
|
["climate.set_power"],
|
|
)
|
|
self.assertTrue(result.uncertain)
|
|
self.assertFalse(result.execution_eligible)
|
|
|
|
def test_unsupported_semantics_never_create_fake_executor_payloads(self) -> None:
|
|
for text in ("打开前大灯", "把主驾座椅调整为零重力模式"):
|
|
with self.subTest(text=text):
|
|
result = self.processor.process(text)
|
|
self.assertEqual(result.semantic_status, "recognized_unsupported")
|
|
self.assertTrue(result.clean_text)
|
|
self.assertTrue(result.commands)
|
|
self.assertTrue(
|
|
all(command.executor_payload is None for command in result.commands)
|
|
)
|
|
|
|
def test_zero_gravity_is_real_passenger_mode_but_not_driver_capability(self) -> None:
|
|
passenger = self.processor.process("打开零重力模式")
|
|
self.assertEqual(passenger.semantic_status, "recognized_executable")
|
|
self.assertEqual(passenger.commands[0].mode_profile_id, "zero_gravity")
|
|
payload = passenger.commands[0].executor_payload
|
|
self.assertIsNotNone(payload)
|
|
self.assertEqual(len(payload["operations"]), 4)
|
|
self.assertEqual(
|
|
{operation["target"] for operation in payload["operations"]},
|
|
{"passenger"},
|
|
)
|
|
|
|
driver = self.processor.process("把主驾座椅调整为零重力模式")
|
|
self.assertEqual(driver.semantic_status, "recognized_unsupported")
|
|
self.assertIsNone(driver.commands[0].executor_payload)
|
|
|
|
def test_native_executor_can_compile_reported_executable_cases(self) -> None:
|
|
for text in (
|
|
"把遮阳帘打开",
|
|
"打开主驾座椅通风",
|
|
"打开一半车窗",
|
|
"打开主驾座椅加热",
|
|
):
|
|
with self.subTest(text=text):
|
|
result = self.processor.process(text)
|
|
payloads = [command.executor_payload for command in result.commands]
|
|
self.assertTrue(all(payload is not None for payload in payloads))
|
|
config = self.native_runtime.build_function_config(
|
|
payloads[0]["domain"], payloads, "speech-alignment", "test"
|
|
)
|
|
plan = self.native_runtime.compile(config)
|
|
self.assertTrue(plan["spec"]["commands"])
|
|
|
|
def test_catalog_driven_resolution_matrix_covers_every_definition(self) -> None:
|
|
definitions = self.resolver.definitions()
|
|
self.assertEqual(len(definitions), 44)
|
|
projections = (
|
|
"canonical_action",
|
|
"target_preservation",
|
|
"parameter_preservation",
|
|
"intent_ir_projection",
|
|
"executor_projection",
|
|
)
|
|
for definition in definitions:
|
|
target = definition.target_scopes[0] if definition.target_scopes else "vehicle"
|
|
parameters = _sample_parameters(definition)
|
|
expected = (
|
|
"unsupported"
|
|
if definition.availability == "catalog_only"
|
|
or (
|
|
definition.availability == "native_can"
|
|
and definition.name not in self.native_runtime.supported_capabilities
|
|
)
|
|
else "executable"
|
|
)
|
|
for projection in projections:
|
|
with self.subTest(capability=definition.name, projection=projection):
|
|
result = self.resolver.resolve(
|
|
domain=definition.domain,
|
|
semantic_property=definition.name,
|
|
target_scope=target,
|
|
target_device=None,
|
|
operation="set",
|
|
parameters=parameters,
|
|
source_text=f"{projection}:{definition.name}:{target}",
|
|
preferred_capability=definition.name,
|
|
)
|
|
self.assertEqual(result.definition.name, definition.name)
|
|
self.assertEqual(result.status, expected)
|
|
if expected == "executable":
|
|
self.assertEqual(result.executor_payload["target"], target)
|
|
self.assertEqual(result.executor_payload["parameters"], parameters)
|
|
self.assertEqual(result.intent_ir_action["target"]["zone"], target)
|
|
else:
|
|
self.assertIsNone(result.executor_payload)
|
|
|
|
def test_vocabulary_does_not_define_vehicle_capabilities(self) -> None:
|
|
self.assertNotIn("capability", DeviceTerm.__dataclass_fields__)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|