vee
77 lines
3.9 KiB
Python
77 lines
3.9 KiB
Python
"""Seat selection for new mode drafts, not for replaying saved user modes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from collections.abc import Mapping, Sequence
|
|
from copy import deepcopy
|
|
from typing import Any
|
|
|
|
from cognitive_agent.application.mode_schedule import ModeSchedule
|
|
from cognitive_agent.application.semantic_ir import resolve_scene_seat_scopes
|
|
|
|
|
|
class ModeCreationSeatPolicyError(ValueError):
|
|
"""A source mode cannot satisfy the selected seat without substitution."""
|
|
|
|
|
|
def creation_seat_scopes(text: str) -> frozenset[str]:
|
|
"""Default to passenger; HVAC/window zone names do not select a seat."""
|
|
normalized = text.replace("主副驾", "主驾和副驾")
|
|
# Do not let the generic 驾驶席/驾驶位 aliases inside 副驾驶席/位
|
|
# accidentally select the driver as well as the passenger.
|
|
normalized = re.sub(r"副驾驶(?:席|位|座(?!椅))?", "副驾", normalized)
|
|
if re.search(r"(?:未知|不明|不存在的?|任意)(?:一个)?(?:座位|座椅)|第[四五六七八九0-9]+排座", normalized):
|
|
return frozenset({"unsupported"})
|
|
normalized = re.sub(r"(?:前排)?两(?:个|张)?(?:座椅|座位|座)|双座(?:椅)?", "主驾和副驾", normalized)
|
|
excluded = set()
|
|
for scope, aliases in (("driver", r"主驾(?:驶)?|驾驶位|驾驶席|司机位"),
|
|
("passenger", r"副驾(?:驶)?|乘客位")):
|
|
pattern = re.compile(
|
|
rf"(?:不要|别|不用|无需|不需要|不涉及|不包含|排除)"
|
|
rf"(?:再|去|动|移动|调节|调整|控制|联动|编排|使用|添加|包含)*"
|
|
rf"(?:{aliases})(?:座椅|座位)?|"
|
|
rf"(?:{aliases})(?:座椅|座位)?(?:不动|不要动|不用动|不联动|不参与|不编排)"
|
|
)
|
|
if pattern.search(normalized):
|
|
excluded.add(scope)
|
|
normalized = pattern.sub("", normalized)
|
|
selected = resolve_scene_seat_scopes(normalized) or frozenset({"passenger"})
|
|
return frozenset(selected - excluded)
|
|
|
|
|
|
def normalized_seat_scope(scope: object) -> str:
|
|
value = str(scope or "")
|
|
return {"front_left": "driver", "front_right": "passenger"}.get(value, value)
|
|
|
|
|
|
def project_seed_operations(
|
|
operations: Sequence[Mapping[str, Any]], scopes: frozenset[str],
|
|
) -> list[dict[str, Any]]:
|
|
"""Choose from trusted seat variants. Never retarget a passenger capability."""
|
|
if "unsupported" in scopes:
|
|
raise ModeCreationSeatPolicyError("当前功能模式仅支持明确的主驾或副驾座椅，不会替换为其他座位。")
|
|
seats = [op for op in operations if op.get("domain") == "seat"]
|
|
if any(normalized_seat_scope(op.get("target_scope")) not in {"driver", "passenger"} for op in seats):
|
|
raise ModeCreationSeatPolicyError("模式包含无法验证的座位范围，请重新生成。")
|
|
projected = [deepcopy(dict(op)) for op in operations
|
|
if op.get("domain") != "seat" or normalized_seat_scope(op.get("target_scope")) in scopes]
|
|
if seats and not any(op.get("domain") == "seat" for op in projected):
|
|
target = "主驾" if scopes == {"driver"} else "副驾" if scopes == {"passenger"} else "所选座位"
|
|
raise ModeCreationSeatPolicyError(f"该模式没有支持{target}的座椅动作，不会把另一座位的能力冒充为{target}能力。")
|
|
return projected
|
|
|
|
|
|
def project_seed_plan(plan: Mapping[str, Any], scopes: frozenset[str]) -> dict[str, Any]:
|
|
projected = deepcopy(dict(plan))
|
|
projected["requests"] = project_seed_operations(plan.get("requests", []), scopes)
|
|
projected["detected_domains"] = list(dict.fromkeys(op["domain"] for op in projected["requests"]))
|
|
return projected
|
|
|
|
|
|
def project_seed_schedule(schedule: ModeSchedule, scopes: frozenset[str]) -> ModeSchedule:
|
|
document = schedule.to_dict()
|
|
for phase in document["phases"]:
|
|
phase["operations"] = project_seed_operations(phase["operations"], scopes)
|
|
return ModeSchedule.from_dict(document)
|