minghanchen
1285 lines
50 KiB
Python
Executable File
1285 lines
50 KiB
Python
Executable File
"""Deterministic fallback parser for supported seat-control vocabulary."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
import re
|
||
import unicodedata
|
||
from copy import deepcopy
|
||
from typing import Any
|
||
|
||
|
||
TARGET_PATTERN = re.compile(
|
||
r"主副驾|前排右侧|右前座椅|右前座|副驾驶座椅|副驾驶座|副驾驶|副驾位|"
|
||
r"乘客座椅|乘客座|乘客位|主驾驶座椅|主驾驶座|主驾驶|驾驶位|驾驶座|"
|
||
r"司机位|司机座|主驾|前排|副驾"
|
||
)
|
||
FEATURE_PATTERN = re.compile(r"通风|加热")
|
||
STATE_PATTERN = re.compile(
|
||
r"设置为|调到|调至|设为|开到|关闭|关掉|停止|取消|打开|开启|启动|开"
|
||
)
|
||
LEVEL_PATTERN = re.compile(r"(最高|最低|[1-5一二三四五高中低])\s*档")
|
||
POSITION_PATTERN_TEMPLATE = (
|
||
r"{alias}(?:位置)?(?:调到|移动到|移到|设为|设置为|调至|移动至|到|为)?"
|
||
r"(?P<position>[+-]?\d{{1,3}}(?:\.\d+)?)%"
|
||
)
|
||
ANGLE_PATTERN_TEMPLATE = (
|
||
r"{alias}(?:夹角|角度)?(?:调到|放倒到|抬到|移动到|移到|设为|设置为|"
|
||
r"调至|到|为)?(?P<angle>\d{{2,3}})(?:度|°)"
|
||
)
|
||
RELATIVE_ANGLE_PATTERN_TEMPLATE = (
|
||
r"{alias}(?:再|继续|接着|然后)?(?P<direction>向前|往前|朝前|前倾|向后|往后|朝后|后仰|"
|
||
r"向上|往上|抬高|抬起|升高|向下|往下|降低|降下|前|后|上|下)"
|
||
r"(?:移动|移|调节|调整|调|倾斜|抬|升|降)?(?P<delta>\d{{1,3}})(?:度|°)"
|
||
)
|
||
RELATIVE_PERCENT_PATTERN_TEMPLATE = (
|
||
r"{alias}(?:再|继续|接着|然后)?(?P<direction>向前|往前|朝前|前移|向后|往后|朝后|后移|"
|
||
r"向上|往上|抬高|抬起|升高|向下|往下|降低|降下|前|后|上|下)"
|
||
r"(?:移动|移|调节|调整|调|倾斜|抬|升|降)?(?P<delta>\d{{1,3}})%"
|
||
)
|
||
TIMED_MOTION_PATTERN_TEMPLATE = (
|
||
r"{alias}(?:持续|保持)?(?:再|继续|接着|然后)?"
|
||
r"(?P<direction>向前|往前|朝前|前移|前倾|向后|往后|朝后|后移|后仰|"
|
||
r"向上|往上|抬高|抬起|升高|向下|往下|降低|降下|前|后|上|下)"
|
||
r"(?:移动|移|调节|调整|调|倾斜|抬|升|降|运行|运动|持续|保持|时长)*"
|
||
r"(?P<duration>\d+(?:\.\d+)?)(?P<unit>毫秒|ms|秒|s|分钟|分|min)"
|
||
)
|
||
ZERO_POSITION_PATTERN = re.compile(r"归零|回零|复位|零位")
|
||
VENTILATION_FEEDBACK_PATTERN = re.compile(r"(?<!不)(?<!加)热|(?<!不)闷|出汗")
|
||
HEATING_FEEDBACK_PATTERN = re.compile(r"(?<!不)冷|(?<!不)凉")
|
||
MILD_PATTERN = re.compile(r"有点|一点|稍微|微微")
|
||
STRONG_PATTERN = re.compile(r"很|太|非常|特别|难受")
|
||
AIRFLOW_CAUSE_PATTERN = re.compile(r"吹|风太大|风吹")
|
||
FIRST_PERSON_PATTERN = re.compile(r"我|自己")
|
||
VEHICLE_SCOPE_PATTERN = re.compile(r"车上|车内|车里|全车|整车|整个车")
|
||
PASSENGER_ONLY_QUESTION = "当前所有座椅能力仅控制副驾，不支持主驾座椅目标。"
|
||
UNSUPPORTED_SEAT_SCOPE_PATTERN = re.compile(
|
||
r"后排|第二排|二排|第三排|三排|左前座椅|左前座|左前|"
|
||
r"右后座椅|右后座|右后|左后座椅|左后座|左后|"
|
||
r"所有座椅|全部座椅|全车座椅"
|
||
)
|
||
UNSUPPORTED_SEAT_SCOPE_QUESTION = (
|
||
"当前台架仅支持副驾座椅，不能把后排、其他方位或全部座椅请求默认到副驾。"
|
||
)
|
||
|
||
TARGETS = {
|
||
"主驾": ["driver"],
|
||
"驾驶位": ["driver"],
|
||
"司机位": ["driver"],
|
||
"主驾驶": ["driver"],
|
||
"主驾驶座": ["driver"],
|
||
"主驾驶座椅": ["driver"],
|
||
"驾驶座": ["driver"],
|
||
"司机座": ["driver"],
|
||
"副驾": ["passenger"],
|
||
"副驾驶": ["passenger"],
|
||
"副驾驶座": ["passenger"],
|
||
"副驾驶座椅": ["passenger"],
|
||
"副驾位": ["passenger"],
|
||
"乘客位": ["passenger"],
|
||
"乘客座": ["passenger"],
|
||
"乘客座椅": ["passenger"],
|
||
"右前座": ["passenger"],
|
||
"右前座椅": ["passenger"],
|
||
"前排右侧": ["passenger"],
|
||
"主副驾": ["driver", "passenger"],
|
||
"前排": ["driver", "passenger"],
|
||
}
|
||
FEATURES = {"通风": "ventilation", "加热": "heating"}
|
||
STATES = {
|
||
"打开": "on",
|
||
"开启": "on",
|
||
"启动": "on",
|
||
"开": "on",
|
||
"开到": "on",
|
||
"调到": "on",
|
||
"调至": "on",
|
||
"设置为": "on",
|
||
"设为": "on",
|
||
"关闭": "off",
|
||
"关掉": "off",
|
||
"停止": "off",
|
||
"取消": "off",
|
||
}
|
||
LEVELS = {
|
||
"一": 1,
|
||
"二": 2,
|
||
"三": 3,
|
||
"四": 4,
|
||
"五": 5,
|
||
"最低": 1,
|
||
"低": 1,
|
||
"中": 2,
|
||
"最高": 3,
|
||
"高": 3,
|
||
}
|
||
MOTION_AXIS_ALIASES = {
|
||
"longitudinal": ("水平", "前后", "纵向", "滑轨"),
|
||
# Internal capability name is retained for compatibility. On the
|
||
# seat-ota CAN0 firmware this board axis is the physical seat-height axis.
|
||
"cushion_tilt": ("座椅高度", "高度", "坐垫倾角", "坐盆", "坐垫"),
|
||
"backrest_recline": ("靠背", "椅背"),
|
||
"legrest_height": ("腿部支撑", "小腿托", "腿托", "脚托"),
|
||
}
|
||
MOTION_ANGLE_RANGES = {
|
||
"backrest_recline": (95, 163),
|
||
"legrest_height": (95, 167),
|
||
}
|
||
CHINESE_PERCENT_PATTERN = re.compile(r"百分之([零一二三四五六七八九十百两\d]+)")
|
||
CHINESE_DEGREE_PATTERN = re.compile(r"([零一二三四五六七八九十百两]+)(度)")
|
||
CHINESE_DURATION_PATTERN = re.compile(
|
||
r"([零一二三四五六七八九十百两]+)(毫秒|秒|分钟|分)"
|
||
)
|
||
RELATIVE_DIRECTION_SIGNS = {
|
||
"longitudinal": {
|
||
"increase": {"向后", "往后", "朝后", "后移", "后"},
|
||
"decrease": {"向前", "往前", "朝前", "前移", "前"},
|
||
},
|
||
"cushion_tilt": {
|
||
"increase": {"向下", "往下", "降低", "降下", "下"},
|
||
"decrease": {"向上", "往上", "抬高", "抬起", "升高", "上"},
|
||
},
|
||
"backrest_recline": {
|
||
"increase": {"向后", "往后", "朝后", "后仰", "后"},
|
||
"decrease": {"向前", "往前", "朝前", "前倾", "前"},
|
||
},
|
||
"legrest_height": {
|
||
"increase": {"向上", "往上", "抬高", "抬起", "升高", "上"},
|
||
"decrease": {"向下", "往下", "降低", "降下", "下"},
|
||
},
|
||
}
|
||
TIMED_MOTION_MIN_DURATION_MS = 300
|
||
TIMED_AXIS_MAX_DURATION_MS = {
|
||
"longitudinal": 30_000,
|
||
"cushion_tilt": 20_000,
|
||
"backrest_recline": 30_000,
|
||
"legrest_height": 25_000,
|
||
}
|
||
TIMED_AXIS_MS_PER_PCT = {
|
||
"longitudinal": 120,
|
||
"cushion_tilt": 80,
|
||
"backrest_recline": 120,
|
||
"legrest_height": 100,
|
||
}
|
||
MOTION_AXIS_LABELS = {
|
||
"longitudinal": "水平轴",
|
||
"cushion_tilt": "座椅高度轴",
|
||
"backrest_recline": "靠背轴",
|
||
"legrest_height": "腿托轴",
|
||
}
|
||
MOTION_ENDPOINT_LABELS = {
|
||
"longitudinal": {"increase": "最后方", "decrease": "最前方"},
|
||
"cushion_tilt": {"increase": "最下方", "decrease": "最上方"},
|
||
"backrest_recline": {"increase": "最后方", "decrease": "最前方"},
|
||
"legrest_height": {"increase": "最上方", "decrease": "最下方"},
|
||
}
|
||
|
||
|
||
def _chinese_integer(token: str) -> int | None:
|
||
if token.isdigit():
|
||
return int(token)
|
||
digits = {
|
||
"零": 0,
|
||
"一": 1,
|
||
"二": 2,
|
||
"两": 2,
|
||
"三": 3,
|
||
"四": 4,
|
||
"五": 5,
|
||
"六": 6,
|
||
"七": 7,
|
||
"八": 8,
|
||
"九": 9,
|
||
}
|
||
total = 0
|
||
current = 0
|
||
for char in token:
|
||
if char in digits:
|
||
current = digits[char]
|
||
elif char == "十":
|
||
total += max(current, 1) * 10
|
||
current = 0
|
||
elif char == "百":
|
||
total += max(current, 1) * 100
|
||
current = 0
|
||
else:
|
||
return None
|
||
return total + current
|
||
|
||
|
||
def _normalize_text(value: str) -> str:
|
||
text = unicodedata.normalize("NFKC", value.strip())
|
||
text = text.replace("半分钟", "0.5分钟").replace("半秒", "0.5秒")
|
||
text = re.sub(r"座椅(?:吹风|送风)", "座椅通风", text)
|
||
text = re.sub(r"座椅(?:取暖|加温)", "座椅加热", text)
|
||
text = text.replace("座椅暖一下", "座椅加热打开")
|
||
|
||
def replace_percent(match: re.Match[str]) -> str:
|
||
number = _chinese_integer(match.group(1))
|
||
return match.group(0) if number is None else f"{number}%"
|
||
|
||
text = CHINESE_PERCENT_PATTERN.sub(replace_percent, text)
|
||
|
||
def replace_degree(match: re.Match[str]) -> str:
|
||
number = _chinese_integer(match.group(1))
|
||
return match.group(0) if number is None else f"{number}{match.group(2)}"
|
||
|
||
text = CHINESE_DEGREE_PATTERN.sub(replace_degree, text)
|
||
|
||
def replace_duration(match: re.Match[str]) -> str:
|
||
number = _chinese_integer(match.group(1))
|
||
return match.group(0) if number is None else f"{number}{match.group(2)}"
|
||
|
||
text = CHINESE_DURATION_PATTERN.sub(replace_duration, text)
|
||
return re.sub(r"\s+", "", text)
|
||
|
||
|
||
def _relative_direction(axis: str, token: str) -> str | None:
|
||
directions = RELATIVE_DIRECTION_SIGNS[axis]
|
||
for operation, tokens in directions.items():
|
||
if token in tokens:
|
||
return operation
|
||
return None
|
||
|
||
|
||
def _context_motion_state(
|
||
context: dict[str, Any] | None, axis: str
|
||
) -> dict[str, Any] | None:
|
||
if not context:
|
||
return None
|
||
state = (
|
||
context.get("devices", {})
|
||
.get("passenger", {})
|
||
.get("position", {})
|
||
.get(axis)
|
||
)
|
||
return state if isinstance(state, dict) else None
|
||
|
||
|
||
def _motor_range(
|
||
context: dict[str, Any] | None, axis: str
|
||
) -> tuple[float, tuple[float, float] | None]:
|
||
motor = (
|
||
(context or {})
|
||
.get("vehicle", {})
|
||
.get("seat", {})
|
||
.get("motors", {})
|
||
.get(axis, {})
|
||
)
|
||
configured_range = motor.get("range") if isinstance(motor, dict) else None
|
||
angle_range = MOTION_ANGLE_RANGES.get(axis)
|
||
if angle_range is not None:
|
||
display_angle = motor.get("display_angle") if isinstance(motor, dict) else None
|
||
if isinstance(display_angle, dict):
|
||
zero = display_angle.get("zero_deg")
|
||
maximum = display_angle.get("max_deg")
|
||
if isinstance(zero, (int, float)) and isinstance(maximum, (int, float)):
|
||
angle_range = (float(zero), float(maximum))
|
||
if isinstance(configured_range, (int, float)) and configured_range > 0:
|
||
return float(configured_range), angle_range
|
||
if angle_range is not None:
|
||
return float(angle_range[1] - angle_range[0]), angle_range
|
||
return 100.0, None
|
||
|
||
|
||
def _clean_number(value: float) -> int | float:
|
||
return int(value) if float(value).is_integer() else round(float(value), 1)
|
||
|
||
|
||
def _duration_ms(value: str, unit: str) -> int:
|
||
multiplier = {
|
||
"毫秒": 1,
|
||
"ms": 1,
|
||
"秒": 1_000,
|
||
"s": 1_000,
|
||
"分钟": 60_000,
|
||
"分": 60_000,
|
||
"min": 60_000,
|
||
}[unit]
|
||
return round(float(value) * multiplier)
|
||
|
||
|
||
def _format_seconds(duration_ms: int) -> str:
|
||
seconds = duration_ms / 1_000
|
||
return f"{seconds:g}"
|
||
|
||
|
||
def _round_half_away_from_zero(value: float) -> int:
|
||
"""Use the integer a user normally expects from a .5 position request."""
|
||
|
||
return math.floor(value + 0.5) if value >= 0 else math.ceil(value - 0.5)
|
||
|
||
|
||
def _format_numeric_request(value: float) -> str:
|
||
return f"{value:g}"
|
||
|
||
|
||
def _timed_motion_action(
|
||
axis: str,
|
||
direction_token: str,
|
||
requested_duration_ms: int,
|
||
context: dict[str, Any] | None,
|
||
) -> tuple[dict[str, Any] | None, str | None]:
|
||
operation = _relative_direction(axis, direction_token)
|
||
if operation is None:
|
||
return None, f"{direction_token}不适用于该座椅轴，请确认移动方向。"
|
||
if requested_duration_ms < 0:
|
||
return None, "座椅按时长移动的时长不能为负数。"
|
||
state = _context_motion_state(context, axis)
|
||
if state is None:
|
||
return None, "缺少当前座椅位置，无法安全计算按时长移动的方向和备用端点。"
|
||
current_pct = state.get("position_pct")
|
||
if (
|
||
not isinstance(current_pct, (int, float))
|
||
or isinstance(current_pct, bool)
|
||
or not 0 <= float(current_pct) <= 100
|
||
):
|
||
return None, "当前座椅位置无效，无法安全执行按时长移动。"
|
||
|
||
maximum_duration_ms = TIMED_AXIS_MAX_DURATION_MS[axis]
|
||
applied_duration_ms = min(
|
||
maximum_duration_ms,
|
||
max(TIMED_MOTION_MIN_DURATION_MS, requested_duration_ms),
|
||
)
|
||
duration_clamped = applied_duration_ms != requested_duration_ms
|
||
estimated_delta_pct = applied_duration_ms / TIMED_AXIS_MS_PER_PCT[axis]
|
||
signed_delta = estimated_delta_pct if operation == "increase" else -estimated_delta_pct
|
||
raw_target = float(current_pct) + signed_delta
|
||
target_pct = round(max(0.0, min(100.0, raw_target)))
|
||
endpoint_expected = None
|
||
if raw_target >= 100:
|
||
endpoint_expected = "far"
|
||
elif raw_target <= 0:
|
||
endpoint_expected = "near"
|
||
|
||
motion_intent: dict[str, Any] = {
|
||
"mode": "timed",
|
||
"direction": operation,
|
||
"start_position_pct": round(float(current_pct)),
|
||
"delta_position_pct": abs(target_pct - round(float(current_pct))),
|
||
"requested_duration_ms": requested_duration_ms,
|
||
"applied_duration_ms": applied_duration_ms,
|
||
"max_duration_ms": maximum_duration_ms,
|
||
"min_duration_ms": TIMED_MOTION_MIN_DURATION_MS,
|
||
"duration_clamped": duration_clamped,
|
||
}
|
||
if endpoint_expected is not None:
|
||
motion_intent["endpoint_expected"] = endpoint_expected
|
||
|
||
action: dict[str, Any] = {
|
||
"target": "passenger",
|
||
"action_type": "position",
|
||
"axis": axis,
|
||
"position_pct": target_pct,
|
||
"motion_intent": motion_intent,
|
||
}
|
||
_, angle_range = _motor_range(context, axis)
|
||
if angle_range is not None:
|
||
target_angle = angle_range[0] + (
|
||
angle_range[1] - angle_range[0]
|
||
) * target_pct / 100
|
||
action["angle_deg"] = _clean_number(target_angle)
|
||
|
||
axis_label = MOTION_AXIS_LABELS[axis]
|
||
if requested_duration_ms > maximum_duration_ms:
|
||
action["user_notice"] = (
|
||
f"{axis_label}最大允许时长为{_format_seconds(maximum_duration_ms)}秒；"
|
||
f"已将用户请求的{_format_seconds(requested_duration_ms)}秒钳制为"
|
||
f"{_format_seconds(applied_duration_ms)}秒并继续执行。"
|
||
)
|
||
elif requested_duration_ms < TIMED_MOTION_MIN_DURATION_MS:
|
||
action["user_notice"] = (
|
||
f"{axis_label}最小允许时长为{_format_seconds(TIMED_MOTION_MIN_DURATION_MS)}秒；"
|
||
f"已按{_format_seconds(applied_duration_ms)}秒继续执行。"
|
||
)
|
||
else:
|
||
action["user_notice"] = (
|
||
f"{axis_label}将按时长执行{_format_seconds(applied_duration_ms)}秒。"
|
||
)
|
||
if endpoint_expected is not None:
|
||
endpoint_label = MOTION_ENDPOINT_LABELS[axis][operation]
|
||
action["user_notice"] += f" 若提前触达机械端点，将在{endpoint_label}停止并按成功返回。"
|
||
return action, None
|
||
|
||
|
||
def _relative_motion_action(
|
||
axis: str,
|
||
direction_token: str,
|
||
delta: int,
|
||
unit: str,
|
||
context: dict[str, Any] | None,
|
||
) -> tuple[dict[str, Any] | None, str | None]:
|
||
operation = _relative_direction(axis, direction_token)
|
||
if operation is None:
|
||
return None, f"{direction_token}不适用于该座椅轴，请确认移动方向。"
|
||
if delta <= 0:
|
||
return None, "座椅相对移动量必须大于0。"
|
||
state = _context_motion_state(context, axis)
|
||
if state is None:
|
||
return None, "缺少当前座椅位置，无法安全计算相对移动目标。"
|
||
current_pct = state.get("position_pct")
|
||
if (
|
||
not isinstance(current_pct, (int, float))
|
||
or isinstance(current_pct, bool)
|
||
or not 0 <= float(current_pct) <= 100
|
||
):
|
||
return None, "当前座椅位置无效，无法安全计算相对移动目标。"
|
||
|
||
physical_range, angle_range = _motor_range(context, axis)
|
||
if unit == "degree" and angle_range is None:
|
||
return None, "该座椅轴没有角度标定，不能按角度执行相对移动。"
|
||
delta_pct = float(delta) if unit == "percent" else delta / physical_range * 100
|
||
signed_delta = delta_pct if operation == "increase" else -delta_pct
|
||
raw_target = float(current_pct) + signed_delta
|
||
clamped = not 0 <= raw_target <= 100
|
||
target_pct = round(max(0.0, min(100.0, raw_target)))
|
||
if target_pct == round(float(current_pct)) and not clamped:
|
||
return None, "相对移动量换算后小于该座椅轴的最小可执行步长。"
|
||
|
||
motion_intent: dict[str, Any] = {
|
||
"mode": "relative",
|
||
"direction": operation,
|
||
"start_position_pct": round(float(current_pct)),
|
||
"delta_position_pct": abs(target_pct - round(float(current_pct))),
|
||
}
|
||
action: dict[str, Any] = {
|
||
"target": "passenger",
|
||
"action_type": "position",
|
||
"axis": axis,
|
||
"position_pct": target_pct,
|
||
"motion_intent": motion_intent,
|
||
}
|
||
if unit == "degree" and angle_range is not None:
|
||
current_angle = state.get("angle_deg")
|
||
if not isinstance(current_angle, (int, float)) or isinstance(
|
||
current_angle, bool
|
||
):
|
||
current_angle = angle_range[0] + (
|
||
angle_range[1] - angle_range[0]
|
||
) * float(current_pct) / 100
|
||
signed_angle = float(delta) if operation == "increase" else -float(delta)
|
||
raw_target_angle = float(current_angle) + signed_angle
|
||
target_angle = max(angle_range[0], min(angle_range[1], raw_target_angle))
|
||
clamped = clamped or target_angle != raw_target_angle
|
||
action["angle_deg"] = _clean_number(target_angle)
|
||
motion_intent["start_angle_deg"] = float(current_angle)
|
||
motion_intent["delta_angle_deg"] = abs(target_angle - float(current_angle))
|
||
if clamped:
|
||
motion_intent["requested_delta_angle_deg"] = float(delta)
|
||
if clamped:
|
||
motion_intent["range_clamped"] = True
|
||
motion_intent["requested_delta_position_pct"] = round(abs(delta_pct))
|
||
endpoint_label = MOTION_ENDPOINT_LABELS[axis][operation]
|
||
requested_value = f"{delta}度" if unit == "degree" else f"{delta}%"
|
||
applied_value = (
|
||
f"{_clean_number(motion_intent.get('delta_angle_deg', 0))}度"
|
||
if unit == "degree"
|
||
else f"{motion_intent['delta_position_pct']}%"
|
||
)
|
||
action["user_notice"] = (
|
||
f"{MOTION_AXIS_LABELS[axis]}{direction_token}移动{requested_value}超过剩余行程；"
|
||
f"已采用备用目标，实际移动最多{applied_value}并到达{endpoint_label}。"
|
||
)
|
||
return action, None
|
||
|
||
|
||
def _nearest_preceding_target(clause: str, position: int) -> str | None:
|
||
candidates = [match for match in TARGET_PATTERN.finditer(clause) if match.start() <= position]
|
||
if candidates:
|
||
return candidates[-1].group(0)
|
||
matches = list(TARGET_PATTERN.finditer(clause))
|
||
return matches[0].group(0) if len(matches) == 1 else None
|
||
|
||
|
||
def _state_for_feature(clause: str, position: int) -> str | None:
|
||
states = list(STATE_PATTERN.finditer(clause))
|
||
if not states:
|
||
return None
|
||
following = [match for match in states if match.start() >= position]
|
||
if following and following[0].start() - position <= 8:
|
||
return STATES[following[0].group(0)]
|
||
if len(states) == 1:
|
||
return STATES[states[0].group(0)]
|
||
preceding = [match for match in states if match.start() < position]
|
||
return STATES[preceding[-1].group(0)] if preceding else None
|
||
|
||
|
||
def _level_for_clause(clause: str) -> int | None:
|
||
match = LEVEL_PATTERN.search(clause)
|
||
if not match:
|
||
return None
|
||
token = match.group(1)
|
||
return int(token) if token.isdigit() else LEVELS[token]
|
||
|
||
|
||
def _context_device(
|
||
context: dict[str, Any] | None, target: str, function: str
|
||
) -> dict[str, Any]:
|
||
if not context:
|
||
return {"state": "off", "level": 0}
|
||
return (
|
||
context.get("devices", {})
|
||
.get(target, {})
|
||
.get(function, {"state": "off", "level": 0})
|
||
)
|
||
|
||
|
||
def _feedback_targets(
|
||
text: str, context: dict[str, Any] | None, parsed: dict[str, Any]
|
||
) -> tuple[list[str], str]:
|
||
matches = list(TARGET_PATTERN.finditer(text))
|
||
if matches:
|
||
targets: list[str] = []
|
||
for match in matches:
|
||
for target in TARGETS[match.group(0)]:
|
||
if target not in targets:
|
||
targets.append(target)
|
||
return targets, "explicit"
|
||
parsed_targets = list(
|
||
dict.fromkeys(
|
||
action.get("target")
|
||
for action in parsed.get("actions", [])
|
||
if action.get("target") in {"driver", "passenger"}
|
||
)
|
||
)
|
||
if parsed_targets:
|
||
return parsed_targets, "model"
|
||
last_target = context.get("last_target") if context else None
|
||
if last_target in {"driver", "passenger"}:
|
||
return [last_target], "history"
|
||
if FIRST_PERSON_PATTERN.search(text):
|
||
return ["passenger"], "passenger_default"
|
||
return ["passenger"], "passenger_default"
|
||
|
||
|
||
def _contains_unsupported_driver_target(text: str) -> bool:
|
||
return any(
|
||
"driver" in TARGETS[match.group(0)]
|
||
for match in TARGET_PATTERN.finditer(text)
|
||
)
|
||
|
||
|
||
def _inferred_level(text: str) -> tuple[int, str]:
|
||
explicit_level = _level_for_clause(text)
|
||
if explicit_level is not None:
|
||
return explicit_level, "explicit"
|
||
if STRONG_PATTERN.search(text):
|
||
return 3, "strong"
|
||
if MILD_PATTERN.search(text):
|
||
return 1, "mild"
|
||
return 2, "normal"
|
||
|
||
|
||
def _decrease_action(
|
||
target: str, function: str, current: dict[str, Any]
|
||
) -> tuple[dict[str, Any], int, int]:
|
||
previous_level = int(current.get("level") or 0)
|
||
if current.get("state") == "on" and previous_level > 1:
|
||
next_level = previous_level - 1
|
||
return (
|
||
{
|
||
"target": target,
|
||
"function": function,
|
||
"state": "on",
|
||
"requested_level": next_level,
|
||
},
|
||
previous_level,
|
||
next_level,
|
||
)
|
||
return (
|
||
{
|
||
"target": target,
|
||
"function": function,
|
||
"state": "off",
|
||
"requested_level": None,
|
||
},
|
||
previous_level,
|
||
0,
|
||
)
|
||
|
||
|
||
def resolve_contextual_intent(
|
||
parsed: dict[str, Any], utterance: str, context: dict[str, Any] | None = None
|
||
) -> dict[str, Any]:
|
||
"""Resolve implicit comfort feedback against the current virtual seat state."""
|
||
|
||
result = deepcopy(parsed)
|
||
text = _normalize_text(utterance)
|
||
ventilation_cues = list(dict.fromkeys(VENTILATION_FEEDBACK_PATTERN.findall(text)))
|
||
cold_cues = list(dict.fromkeys(HEATING_FEEDBACK_PATTERN.findall(text)))
|
||
if not ventilation_cues and not cold_cues:
|
||
return result
|
||
|
||
if _contains_unsupported_driver_target(text):
|
||
result["actions"] = []
|
||
result["ambiguities"] = [
|
||
{"field": "target", "question": PASSENGER_ONLY_QUESTION}
|
||
]
|
||
result["decision_summary"] = [
|
||
"识别到主驾或包含主驾的座椅范围；当前系统不会将其重定向到副驾。"
|
||
]
|
||
result["confidence"] = max(float(result.get("confidence", 0)), 0.99)
|
||
return result
|
||
|
||
if VEHICLE_SCOPE_PATTERN.search(text) and not TARGET_PATTERN.search(text):
|
||
result["actions"] = []
|
||
result["ambiguities"] = [
|
||
{
|
||
"field": "domain",
|
||
"question": "该请求作用于整车舱空气，需要由空调/新风域执行；当前座椅单域不能代替执行。",
|
||
}
|
||
]
|
||
result["decision_summary"] = [
|
||
"“车上/车内”表示整车舱范围，不是主驾或副驾单座位。",
|
||
"“太闷”属于舱内空气质量/新风需求，应路由到 climate 域。",
|
||
"当前未接入空调执行器，未将整车请求错误下发到任一座椅。",
|
||
]
|
||
result["detected_domains"] = ["climate"]
|
||
result["scope"] = "vehicle"
|
||
result["confidence"] = max(float(result.get("confidence", 0)), 0.96)
|
||
return result
|
||
|
||
targets, target_source = _feedback_targets(text, context, result)
|
||
if not targets:
|
||
if not any(item.get("field") == "target" for item in result["ambiguities"]):
|
||
result["ambiguities"].append(
|
||
{"field": "target", "question": "该舒适反馈需要作用于主驾还是副驾？"}
|
||
)
|
||
return result
|
||
|
||
labels = {
|
||
"driver": "主驾",
|
||
"passenger": "副驾",
|
||
"ventilation": "通风",
|
||
"heating": "加热",
|
||
}
|
||
summaries: list[str] = []
|
||
if target_source == "history":
|
||
summaries.append(f"本轮省略了座位，沿用上一轮的{labels[targets[0]]}目标。")
|
||
elif target_source == "passenger_default":
|
||
summaries.append("当前座椅域仅控制副驾，本轮省略座位并安全限定为副驾。")
|
||
|
||
planned_actions: list[dict[str, Any]] = []
|
||
desired_level, intensity = _inferred_level(text)
|
||
airflow_is_cause = bool(AIRFLOW_CAUSE_PATTERN.search(text))
|
||
is_cold = bool(cold_cues)
|
||
is_hot = any(cue in {"热", "出汗"} for cue in ventilation_cues)
|
||
for target in targets:
|
||
ventilation = _context_device(context, target, "ventilation")
|
||
heating = _context_device(context, target, "heating")
|
||
cue_text = "、".join(cold_cues if is_cold else ventilation_cues)
|
||
summaries.append(f"检测到{labels[target]}“{cue_text}”舒适反馈，开始核对当前执行器状态。")
|
||
|
||
if cold_cues and ventilation_cues and not airflow_is_cause:
|
||
balanced_level = min(desired_level, 3)
|
||
planned_actions.extend(
|
||
[
|
||
{
|
||
"target": target,
|
||
"function": "heating",
|
||
"state": "on",
|
||
"requested_level": balanced_level,
|
||
},
|
||
{
|
||
"target": target,
|
||
"function": "ventilation",
|
||
"state": "on",
|
||
"requested_level": balanced_level,
|
||
},
|
||
]
|
||
)
|
||
summaries[-1] = (
|
||
f"检测到{labels[target]}“{'、'.join(cold_cues + ventilation_cues)}”两个并存的舒适维度，"
|
||
"不将其判定为语义冲突。"
|
||
)
|
||
summaries.append(
|
||
f"冷感由{labels[target]}加热{balanced_level}档补偿，闷感由"
|
||
f"{labels[target]}通风{balanced_level}档缓解；生成可审计的组合目标。"
|
||
)
|
||
continue
|
||
|
||
if is_cold and (airflow_is_cause or ventilation.get("state") == "on"):
|
||
action, previous_level, next_level = _decrease_action(
|
||
target, "ventilation", ventilation
|
||
)
|
||
planned_actions.append(action)
|
||
if airflow_is_cause:
|
||
summaries.append(
|
||
f"“{labels[target]}吹得冷”明确将{labels[target]}通风指定为致冷来源，不映射为加热请求。"
|
||
)
|
||
if previous_level > 1:
|
||
summaries.append(
|
||
f"当前{labels[target]}通风为{previous_level}档，先降低到{next_level}档，不同时调整加热。"
|
||
)
|
||
elif previous_level == 1:
|
||
summaries.append(
|
||
f"当前{labels[target]}通风为1档，冷感更可能由已开启通风造成；"
|
||
"先关闭通风，不立即开启加热。"
|
||
)
|
||
else:
|
||
summaries.append(
|
||
f"当前状态显示{labels[target]}通风已关闭，保持关闭；不会因“吹”反向调整加热。"
|
||
)
|
||
continue
|
||
|
||
if is_cold:
|
||
current_level = int(heating.get("level") or 0)
|
||
if heating.get("state") == "on" and current_level > 0:
|
||
next_level = min(max(current_level + 1, desired_level), 3)
|
||
planned_actions.append(
|
||
{
|
||
"target": target,
|
||
"function": "heating",
|
||
"state": "on",
|
||
"requested_level": next_level,
|
||
}
|
||
)
|
||
summaries.append(
|
||
f"{labels[target]}通风已关闭，加热已开启{current_level}档但仍感到冷，"
|
||
f"将加热提升到{next_level}档。"
|
||
)
|
||
else:
|
||
planned_actions.append(
|
||
{
|
||
"target": target,
|
||
"function": "heating",
|
||
"state": "on",
|
||
"requested_level": desired_level,
|
||
}
|
||
)
|
||
summaries.append(
|
||
f"{labels[target]}通风已关闭且加热未开启，冷感仍存在；"
|
||
f"按{'轻微' if intensity == 'mild' else '当前'}强度开启加热{desired_level}档。"
|
||
)
|
||
continue
|
||
|
||
if is_hot and heating.get("state") == "on":
|
||
action, previous_level, next_level = _decrease_action(target, "heating", heating)
|
||
planned_actions.append(action)
|
||
summaries.append(
|
||
f"当前{labels[target]}加热为{previous_level}档，热感优先视为已开启加热的结果；"
|
||
f"先{'关闭加热' if next_level == 0 else f'降低到{next_level}档'}，不立即开启通风。"
|
||
)
|
||
continue
|
||
|
||
current_level = int(ventilation.get("level") or 0)
|
||
if ventilation.get("state") == "on" and current_level > 0:
|
||
next_level = min(max(current_level + 1, desired_level), 3)
|
||
summaries.append(
|
||
f"{labels[target]}通风已开启{current_level}档但舒适问题仍存在，提升到{next_level}档。"
|
||
)
|
||
else:
|
||
next_level = desired_level
|
||
summaries.append(
|
||
f"{labels[target]}没有已开启的加热致因，按当前强度开启通风{next_level}档。"
|
||
)
|
||
planned_actions.append(
|
||
{
|
||
"target": target,
|
||
"function": "ventilation",
|
||
"state": "on",
|
||
"requested_level": next_level,
|
||
}
|
||
)
|
||
|
||
result["actions"] = planned_actions
|
||
result["ambiguities"] = [
|
||
item
|
||
for item in result["ambiguities"]
|
||
if item.get("field") not in {"capability", "state", "target"}
|
||
]
|
||
result["decision_summary"] = summaries
|
||
result["confidence"] = max(float(result.get("confidence", 0)), 0.9)
|
||
return result
|
||
|
||
|
||
def parse_local_intent(
|
||
utterance: str, context: dict[str, Any] | None = None
|
||
) -> dict[str, Any]:
|
||
"""Parse common Chinese seat commands without making hardware assumptions."""
|
||
|
||
text = _normalize_text(utterance)
|
||
clauses = [part for part in re.split(r"[，,。；;！!？?]", text) if part]
|
||
actions: list[dict[str, Any]] = []
|
||
motion_actions: list[dict[str, Any]] = []
|
||
ambiguities: list[dict[str, str]] = []
|
||
|
||
unsupported_scope = UNSUPPORTED_SEAT_SCOPE_PATTERN.search(text)
|
||
if unsupported_scope is not None:
|
||
return {
|
||
"actions": [],
|
||
"ambiguities": [
|
||
{
|
||
"field": "target",
|
||
"question": UNSUPPORTED_SEAT_SCOPE_QUESTION,
|
||
}
|
||
],
|
||
"decision_summary": [
|
||
f"识别到未支持的座椅目标范围“{unsupported_scope.group(0)}”，未生成副驾动作。"
|
||
],
|
||
"confidence": 0.99,
|
||
}
|
||
|
||
mentioned_axes = [
|
||
axis
|
||
for axis, aliases in MOTION_AXIS_ALIASES.items()
|
||
if any(alias in text for alias in aliases)
|
||
]
|
||
if ZERO_POSITION_PATTERN.search(text) and (mentioned_axes or "四轴" in text):
|
||
reset_axes = list(MOTION_AXIS_ALIASES) if "四轴" in text else mentioned_axes
|
||
if _contains_unsupported_driver_target(text):
|
||
ambiguities.append(
|
||
{"field": "target", "question": PASSENGER_ONLY_QUESTION}
|
||
)
|
||
else:
|
||
motion_actions.extend(
|
||
{
|
||
"target": "passenger",
|
||
"action_type": "position",
|
||
"axis": axis,
|
||
"position_pct": 0,
|
||
}
|
||
for axis in reset_axes
|
||
)
|
||
|
||
for clause in clauses:
|
||
level = _level_for_clause(clause)
|
||
clause_states = {
|
||
STATES[match.group(0)] for match in STATE_PATTERN.finditer(clause)
|
||
}
|
||
feature_matches = list(FEATURE_PATTERN.finditer(clause))
|
||
contradictory_feature_state = len(clause_states) > 1 and bool(feature_matches)
|
||
if contradictory_feature_state:
|
||
ambiguities.append(
|
||
{
|
||
"field": "passenger.thermal.state",
|
||
"question": "同一条指令同时包含打开和关闭，请确认座椅功能的最终状态。",
|
||
}
|
||
)
|
||
for feature_match in feature_matches:
|
||
if contradictory_feature_state:
|
||
continue
|
||
feature_zh = feature_match.group(0)
|
||
target_zh = _nearest_preceding_target(clause, feature_match.start())
|
||
state = _state_for_feature(clause, feature_match.end())
|
||
if target_zh is None:
|
||
target_zh = "副驾"
|
||
if "driver" in TARGETS[target_zh]:
|
||
ambiguities.append(
|
||
{"field": "target", "question": PASSENGER_ONLY_QUESTION}
|
||
)
|
||
continue
|
||
if level is not None and level > 3:
|
||
ambiguities.append(
|
||
{
|
||
"field": f"passenger.{FEATURES[feature_zh]}.level",
|
||
"question": f"{feature_zh}档位必须在1到3档之间。",
|
||
}
|
||
)
|
||
continue
|
||
if state is None and level is not None:
|
||
state = "on"
|
||
if state is None:
|
||
ambiguities.append(
|
||
{
|
||
"field": "state",
|
||
"question": f"请确认{target_zh}{feature_zh}是打开还是关闭。",
|
||
}
|
||
)
|
||
continue
|
||
requested_level = level
|
||
if state == "on" and requested_level is None:
|
||
requested_level = 1
|
||
for target in TARGETS[target_zh]:
|
||
actions.append(
|
||
{
|
||
"target": target,
|
||
"function": FEATURES[feature_zh],
|
||
"state": state,
|
||
"requested_level": requested_level,
|
||
}
|
||
)
|
||
|
||
if ZERO_POSITION_PATTERN.search(text):
|
||
continue
|
||
|
||
clause_axes = [
|
||
axis
|
||
for axis, aliases in MOTION_AXIS_ALIASES.items()
|
||
if any(alias in clause for alias in aliases)
|
||
]
|
||
percentages = list(
|
||
re.finditer(r"(?P<position>[+-]?\d{1,3}(?:\.\d+)?)%", clause)
|
||
)
|
||
shared_position = None
|
||
if len(clause_axes) > 1 and len(percentages) == 1 and re.search(
|
||
r"都|全部|均", clause
|
||
):
|
||
requested_shared_value = float(percentages[0].group("position"))
|
||
rounded_shared_position = _round_half_away_from_zero(
|
||
requested_shared_value
|
||
)
|
||
shared_position = max(0, min(100, rounded_shared_position))
|
||
target_zh = (
|
||
_nearest_preceding_target(clause, percentages[0].start()) or "副驾"
|
||
)
|
||
if "driver" in TARGETS[target_zh]:
|
||
ambiguities.append(
|
||
{"field": "target", "question": PASSENGER_ONLY_QUESTION}
|
||
)
|
||
else:
|
||
motion_actions.extend(
|
||
{
|
||
"target": "passenger",
|
||
"action_type": "position",
|
||
"axis": axis,
|
||
"position_pct": shared_position,
|
||
**(
|
||
{
|
||
"user_notice": (
|
||
f"{MOTION_AXIS_LABELS[axis]}请求位置"
|
||
f"{_format_numeric_request(requested_shared_value)}%；"
|
||
+ (
|
||
f"已取整为{rounded_shared_position}%。"
|
||
if requested_shared_value
|
||
!= rounded_shared_position
|
||
else ""
|
||
)
|
||
+ (
|
||
f" 取整后超过0到100%行程，已采用"
|
||
f"{shared_position}%备用目标并移动到对应机械末端。"
|
||
if rounded_shared_position != shared_position
|
||
else ""
|
||
)
|
||
)
|
||
}
|
||
if requested_shared_value != shared_position
|
||
else {}
|
||
),
|
||
}
|
||
for axis in clause_axes
|
||
)
|
||
|
||
if shared_position is not None:
|
||
continue
|
||
|
||
timed_axes: set[str] = set()
|
||
timed_aliases = {
|
||
**MOTION_AXIS_ALIASES,
|
||
"longitudinal": (*MOTION_AXIS_ALIASES["longitudinal"], "座椅", "座位"),
|
||
}
|
||
for axis, aliases in timed_aliases.items():
|
||
for alias in aliases:
|
||
match = re.search(
|
||
TIMED_MOTION_PATTERN_TEMPLATE.format(alias=re.escape(alias)),
|
||
clause,
|
||
flags=re.IGNORECASE,
|
||
)
|
||
if match is None:
|
||
continue
|
||
timed_axes.add(axis)
|
||
target_zh = _nearest_preceding_target(clause, match.start()) or "副驾"
|
||
if "driver" in TARGETS[target_zh]:
|
||
ambiguities.append(
|
||
{"field": "target", "question": PASSENGER_ONLY_QUESTION}
|
||
)
|
||
break
|
||
action, question = _timed_motion_action(
|
||
axis,
|
||
match.group("direction"),
|
||
_duration_ms(match.group("duration"), match.group("unit").lower()),
|
||
context,
|
||
)
|
||
if action is None:
|
||
ambiguities.append(
|
||
{
|
||
"field": f"passenger.position.{axis}",
|
||
"question": question or "无法计算座椅按时长移动目标。",
|
||
}
|
||
)
|
||
else:
|
||
motion_actions.append(action)
|
||
break
|
||
|
||
relative_axes: set[str] = set()
|
||
for axis, aliases in MOTION_AXIS_ALIASES.items():
|
||
if axis in timed_axes:
|
||
continue
|
||
for alias in aliases:
|
||
angle_match = re.search(
|
||
RELATIVE_ANGLE_PATTERN_TEMPLATE.format(alias=re.escape(alias)),
|
||
clause,
|
||
)
|
||
percent_match = re.search(
|
||
RELATIVE_PERCENT_PATTERN_TEMPLATE.format(alias=re.escape(alias)),
|
||
clause,
|
||
)
|
||
match = angle_match or percent_match
|
||
if match is None:
|
||
continue
|
||
relative_axes.add(axis)
|
||
target_zh = _nearest_preceding_target(clause, match.start()) or "副驾"
|
||
if "driver" in TARGETS[target_zh]:
|
||
ambiguities.append(
|
||
{"field": "target", "question": PASSENGER_ONLY_QUESTION}
|
||
)
|
||
break
|
||
unit = "degree" if angle_match is not None else "percent"
|
||
action, question = _relative_motion_action(
|
||
axis,
|
||
match.group("direction"),
|
||
int(match.group("delta")),
|
||
unit,
|
||
context,
|
||
)
|
||
if action is None:
|
||
ambiguities.append(
|
||
{
|
||
"field": f"passenger.position.{axis}",
|
||
"question": question or "无法计算座椅相对移动目标。",
|
||
}
|
||
)
|
||
else:
|
||
motion_actions.append(action)
|
||
break
|
||
|
||
endpoint_actions: dict[str, int] = {}
|
||
has_specific_non_longitudinal_axis = any(
|
||
any(alias in clause for alias in MOTION_AXIS_ALIASES[axis])
|
||
for axis in ("cushion_tilt", "backrest_recline", "legrest_height")
|
||
)
|
||
has_longitudinal_axis = any(
|
||
alias in clause for alias in MOTION_AXIS_ALIASES["longitudinal"]
|
||
) or (
|
||
not has_specific_non_longitudinal_axis
|
||
and any(alias in clause for alias in ("座椅", "座位"))
|
||
)
|
||
if has_longitudinal_axis:
|
||
if re.search(r"最前面|最前|向前移到底|推到最前", clause):
|
||
endpoint_actions["longitudinal"] = 0
|
||
elif re.search(r"最后面|最后|向后移到底|移到最后", clause):
|
||
endpoint_actions["longitudinal"] = 100
|
||
cushion_alias = r"(?:座椅高度|高度|坐垫倾角|坐盆|坐垫)"
|
||
if re.search(
|
||
cushion_alias + r"(?:抬到最上方|调到最上方|移到最上方|最高|抬到头|抬到底)",
|
||
clause,
|
||
):
|
||
endpoint_actions["cushion_tilt"] = 0
|
||
elif re.search(
|
||
cushion_alias + r"(?:降到最下方|调到最下方|移到最下方|最低|降到头|降到底)",
|
||
clause,
|
||
):
|
||
endpoint_actions["cushion_tilt"] = 100
|
||
backrest_alias = r"(?:靠背|椅背)"
|
||
if re.search(backrest_alias + r"(?:调直|立直|竖直)", clause):
|
||
endpoint_actions["backrest_recline"] = 0
|
||
elif re.search(
|
||
backrest_alias
|
||
+ r"(?:放倒到最后|放倒到底|完全放倒|后仰到底|向后移到底|最大角度)",
|
||
clause,
|
||
):
|
||
endpoint_actions["backrest_recline"] = 100
|
||
legrest_alias = r"(?:腿部支撑|小腿托|腿托|脚托)"
|
||
if re.search(
|
||
legrest_alias
|
||
+ r"(?:收起|收回|降到最下方|调到最下方|降到最低|最低|降到头|降到底)",
|
||
clause,
|
||
):
|
||
endpoint_actions["legrest_height"] = 0
|
||
elif re.search(
|
||
legrest_alias
|
||
+ r"(?:升到最上方|抬到最上方|调到最上方|升到最高|抬到最高|最高|完全抬起|升到头|升到底|抬到底)",
|
||
clause,
|
||
):
|
||
endpoint_actions["legrest_height"] = 100
|
||
for axis, position_pct in endpoint_actions.items():
|
||
target_zh = _nearest_preceding_target(clause, len(clause)) or "副驾"
|
||
if "driver" in TARGETS[target_zh]:
|
||
ambiguities.append(
|
||
{"field": "target", "question": PASSENGER_ONLY_QUESTION}
|
||
)
|
||
continue
|
||
motion_actions.append(
|
||
{
|
||
"target": "passenger",
|
||
"action_type": "position",
|
||
"axis": axis,
|
||
"position_pct": position_pct,
|
||
}
|
||
)
|
||
|
||
for axis, aliases in MOTION_AXIS_ALIASES.items():
|
||
if axis in endpoint_actions or axis in relative_axes or axis in timed_axes:
|
||
continue
|
||
for alias in aliases:
|
||
position_match = re.search(
|
||
POSITION_PATTERN_TEMPLATE.format(alias=re.escape(alias)),
|
||
clause,
|
||
)
|
||
angle_match = None
|
||
if axis in MOTION_ANGLE_RANGES:
|
||
angle_match = re.search(
|
||
ANGLE_PATTERN_TEMPLATE.format(alias=re.escape(alias)),
|
||
clause,
|
||
)
|
||
match = position_match or angle_match
|
||
if match is None:
|
||
continue
|
||
target_zh = _nearest_preceding_target(clause, match.start()) or "副驾"
|
||
if "driver" in TARGETS[target_zh]:
|
||
ambiguities.append(
|
||
{"field": "target", "question": PASSENGER_ONLY_QUESTION}
|
||
)
|
||
break
|
||
absolute_angle = None
|
||
if position_match is not None:
|
||
requested_position_value = float(
|
||
position_match.group("position")
|
||
)
|
||
requested_position_pct = _round_half_away_from_zero(
|
||
requested_position_value
|
||
)
|
||
position_pct = max(0, min(100, requested_position_pct))
|
||
else:
|
||
requested_angle = int(angle_match.group("angle"))
|
||
minimum, maximum = MOTION_ANGLE_RANGES[axis]
|
||
angle = max(minimum, min(maximum, requested_angle))
|
||
position_pct = round(
|
||
(angle - minimum) * 100 / (maximum - minimum)
|
||
)
|
||
absolute_angle = angle
|
||
motion_action = {
|
||
"target": "passenger",
|
||
"action_type": "position",
|
||
"axis": axis,
|
||
"position_pct": position_pct,
|
||
}
|
||
if absolute_angle is not None:
|
||
motion_action["angle_deg"] = absolute_angle
|
||
if requested_angle != absolute_angle:
|
||
endpoint_label = (
|
||
MOTION_ENDPOINT_LABELS[axis]["increase"]
|
||
if requested_angle > maximum
|
||
else MOTION_ENDPOINT_LABELS[axis]["decrease"]
|
||
)
|
||
motion_action["user_notice"] = (
|
||
f"{MOTION_AXIS_LABELS[axis]}请求角度{requested_angle}度超过"
|
||
f"{minimum}到{maximum}度行程；已采用备用目标"
|
||
f"{absolute_angle}度并移动到{endpoint_label}。"
|
||
)
|
||
else:
|
||
notices: list[str] = []
|
||
if requested_position_value != requested_position_pct:
|
||
notices.append(
|
||
f"{MOTION_AXIS_LABELS[axis]}请求位置"
|
||
f"{_format_numeric_request(requested_position_value)}%；"
|
||
f"已取整为{requested_position_pct}%。"
|
||
)
|
||
if requested_position_pct != position_pct:
|
||
notices.append(
|
||
f"取整后超过0到100%行程；已采用{position_pct}%的备用目标"
|
||
"并移动到对应机械末端。"
|
||
)
|
||
if notices:
|
||
motion_action["user_notice"] = " ".join(notices)
|
||
motion_actions.append(motion_action)
|
||
break
|
||
|
||
merged: dict[tuple[str, str], dict[str, Any]] = {}
|
||
for action in actions:
|
||
key = (action["target"], action["function"])
|
||
previous = merged.get(key)
|
||
if previous and previous["state"] != action["state"]:
|
||
ambiguities.append(
|
||
{
|
||
"field": f"{key[0]}.{key[1]}.state",
|
||
"question": "同一个座椅功能同时出现打开和关闭指令，请确认最终状态。",
|
||
}
|
||
)
|
||
merged[key] = action
|
||
merged_motion: dict[tuple[str, str], dict[str, Any]] = {}
|
||
for action in motion_actions:
|
||
key = (action["target"], action["axis"])
|
||
previous = merged_motion.get(key)
|
||
if previous and previous["position_pct"] != action["position_pct"]:
|
||
ambiguities.append(
|
||
{
|
||
"field": f"{key[0]}.position.{key[1]}",
|
||
"question": "同一个座椅轴出现多个位置目标，请确认最终位置。",
|
||
}
|
||
)
|
||
merged_motion[key] = action
|
||
actions = [*merged.values(), *merged_motion.values()]
|
||
|
||
parsed = resolve_contextual_intent(
|
||
{
|
||
"actions": actions,
|
||
"ambiguities": ambiguities,
|
||
"decision_summary": [],
|
||
"confidence": 0.96 if actions and not ambiguities else 0.45,
|
||
},
|
||
utterance,
|
||
context,
|
||
)
|
||
actions = parsed["actions"]
|
||
ambiguities = parsed["ambiguities"]
|
||
|
||
if (
|
||
not actions
|
||
and _contains_unsupported_driver_target(text)
|
||
and not any(item.get("field") == "target" for item in ambiguities)
|
||
):
|
||
ambiguities.append(
|
||
{"field": "target", "question": PASSENGER_ONLY_QUESTION}
|
||
)
|
||
|
||
if not actions and not ambiguities:
|
||
ambiguities.append(
|
||
{
|
||
"field": "capability",
|
||
"question": "当前仅支持副驾座椅通风、加热、移动和姿态预设能力。",
|
||
}
|
||
)
|
||
|
||
summary = ["识别为副驾座椅功能单域请求。", *parsed["decision_summary"]]
|
||
labels = {
|
||
"driver": "主驾",
|
||
"passenger": "副驾",
|
||
"ventilation": "通风",
|
||
"heating": "加热",
|
||
"on": "打开",
|
||
"off": "关闭",
|
||
}
|
||
axis_labels = {
|
||
"longitudinal": "水平",
|
||
"cushion_tilt": "高度",
|
||
"backrest_recline": "靠背",
|
||
"legrest_height": "腿托",
|
||
}
|
||
for action in actions:
|
||
if action.get("action_type") == "position":
|
||
summary.append(
|
||
f"目标：副驾座椅{axis_labels[action['axis']]}位置"
|
||
f"{action['position_pct']}%。"
|
||
)
|
||
if action.get("user_notice"):
|
||
summary.append(str(action["user_notice"]))
|
||
else:
|
||
summary.append(
|
||
f"目标：{labels[action['target']]}座椅"
|
||
f"{labels[action['function']]}{labels[action['state']]}。"
|
||
)
|
||
if any(action.get("requested_level") is not None for action in actions):
|
||
summary.append("目标档位已写入功能配置；现有 CAN 矩阵仅支持档位虚拟回读。")
|
||
|
||
response = {
|
||
"actions": actions,
|
||
"ambiguities": ambiguities,
|
||
"decision_summary": summary,
|
||
"confidence": max(parsed["confidence"], 0.96) if actions and not ambiguities else 0.45,
|
||
}
|
||
for field in ("detected_domains", "scope"):
|
||
if field in parsed:
|
||
response[field] = parsed[field]
|
||
return response
|