backend

src/cognitive_agent/application/model_plan_adapter.py

@@ -10,6 +10,7 @@ from cognitive_agent.application.seat_calibration import (
     ANGLE_CALIBRATIONS,
     angle_to_position_pct,
 )
+from cognitive_agent.application.vehicle_profile import VehicleProfile
 
 
 SEAT_CAPABILITIES = {
@@ -23,6 +24,7 @@ SEAT_MOTION_CAPABILITIES = {
     "seat.set_legrest_height": "legrest_height",
     "seat.set_legrest_extension": "legrest_extension",
 }
+SEAT_RELATIVE_PHYSICAL_PARAMETERS = {"distance_mm", "angle_delta_deg"}
 SAVE_PRESET_CAPABILITY = "seat.save_posture_preset"
 APPLY_PRESET_CAPABILITY = "seat.apply_posture_preset"
 HEX_COLOR = re.compile(r"^#[0-9A-Fa-f]{6}$")
@@ -45,6 +47,7 @@ class ModelPlanAdapter:
         self,
         plan: dict[str, Any],
         context: dict[str, Any] | None = None,
+        vehicle_profile: VehicleProfile | None = None,
     ) -> dict[str, Any]:
         actions: list[dict[str, Any]] = []
         domain_actions: list[dict[str, Any]] = []
@@ -73,6 +76,11 @@ class ModelPlanAdapter:
             }
             allowed_parameters = {item["name"] for item in capability.get("parameters", [])}
             unknown_parameters = set(parameters) - allowed_parameters
+            if (
+                domain == "seat"
+                and capability_name in SEAT_MOTION_CAPABILITIES
+            ):
+                unknown_parameters -= SEAT_RELATIVE_PHYSICAL_PARAMETERS
             if unknown_parameters:
                 ambiguities.append(
                     {
@@ -162,6 +170,7 @@ class ModelPlanAdapter:
                         operation_parameters,
                         ambiguities,
                         context,
+                        vehicle_profile,
                     )
                     if action is not None:
                         actions.append(action)
@@ -174,7 +183,7 @@ class ModelPlanAdapter:
                 )
                 continue
             action = self._seat_action(
-                index, request, parameters, ambiguities, context
+                index, request, parameters, ambiguities, context, vehicle_profile
             )
             if action is not None:
                 actions.append(action)
@@ -367,6 +376,7 @@ class ModelPlanAdapter:
         parameters: dict[str, Any],
         ambiguities: list[dict[str, str]],
         context: dict[str, Any],
+        vehicle_profile: VehicleProfile | None = None,
     ) -> dict[str, Any] | None:
         target = request.get("target_scope")
         if target not in {"driver", "passenger"}:
@@ -381,21 +391,68 @@ class ModelPlanAdapter:
         capability = request["capability"]
         if capability in SEAT_MOTION_CAPABILITIES:
             axis = SEAT_MOTION_CAPABILITIES[capability]
+            physical_action = ModelPlanAdapter._relative_physical_motion_action(
+                index,
+                request,
+                parameters,
+                ambiguities,
+                context,
+                target,
+                axis,
+                vehicle_profile,
+            )
+            if physical_action is not None:
+                return physical_action
             angle = parameters.get("angle_deg")
-            if angle is not None and capability in ANGLE_CALIBRATIONS:
-                calibration = ANGLE_CALIBRATIONS[capability]
-                if (
-                    not isinstance(angle, (int, float))
-                    or isinstance(angle, bool)
-                    or not calibration.minimum_deg <= float(angle) <= calibration.maximum_deg
-                ):
+            if angle is not None:
+                if not isinstance(angle, (int, float)) or isinstance(angle, bool):
                     ambiguities.append(
                         {
                             "field": f"requests[{index}].parameters.angle_deg",
-                            "question": (
-                                f"{capability} 的夹角必须在 "
-                                f"{calibration.minimum_deg:g}-{calibration.maximum_deg:g} 度之间。"
-                            ),
+                            "question": f"{capability} 的夹角必须是数字。",
+                        }
+                    )
+                    return None
+                profile_motor = (
+                    vehicle_profile.seat_motors.get(axis)
+                    if vehicle_profile is not None
+                    else None
+                )
+                if profile_motor is not None and profile_motor.display_angle:
+                    if not profile_motor.contains_display_angle(float(angle)):
+                        display_angle = profile_motor.display_angle
+                        ambiguities.append(
+                            {
+                                "field": f"requests[{index}].parameters.angle_deg",
+                                "question": (
+                                    f"{capability} 的夹角必须在 "
+                                    f"{display_angle['zero_deg']:g}-{display_angle['max_deg']:g} 度之间。"
+                                ),
+                            }
+                        )
+                        return None
+                    position_pct = profile_motor.position_pct_for_display_angle(
+                        float(angle)
+                    )
+                elif capability in ANGLE_CALIBRATIONS:
+                    calibration = ANGLE_CALIBRATIONS[capability]
+                    if not calibration.minimum_deg <= float(angle) <= calibration.maximum_deg:
+                        ambiguities.append(
+                            {
+                                "field": f"requests[{index}].parameters.angle_deg",
+                                "question": (
+                                    f"{capability} 的夹角必须在 "
+                                    f"{calibration.minimum_deg:g}-{calibration.maximum_deg:g} 度之间。"
+                                ),
+                            }
+                        )
+                        return None
+                    position_pct = angle_to_position_pct(capability, float(angle))
+                else:
+                    ambiguities.append(
+                        {
+                            "field": f"requests[{index}].parameters.position_pct",
+                            "question": "当前车型未给出该座椅轴的角度标定，需要 0-100 的绝对百分比。",
                         }
                     )
                     return None
@@ -406,7 +463,7 @@ class ModelPlanAdapter:
                     "action_type": "position",
                     "target": target,
                     "axis": axis,
-                    "position_pct": angle_to_position_pct(capability, float(angle)),
+                    "position_pct": position_pct,
                     "angle_deg": exact_angle,
                 }
                 return ModelPlanAdapter._preserve_relative_motion(
@@ -460,6 +517,157 @@ class ModelPlanAdapter:
             ),
         }
 
+    @staticmethod
+    def _relative_physical_motion_action(
+        index: int,
+        request: dict[str, Any],
+        parameters: dict[str, Any],
+        ambiguities: list[dict[str, str]],
+        context: dict[str, Any],
+        target: str,
+        axis: str,
+        vehicle_profile: VehicleProfile | None,
+    ) -> dict[str, Any] | None:
+        physical_parameters = [
+            name
+            for name in ("distance_mm", "angle_delta_deg")
+            if parameters.get(name) is not None
+        ]
+        if not physical_parameters:
+            return None
+        if len(physical_parameters) > 1:
+            ambiguities.append(
+                {
+                    "field": f"requests[{index}].parameters",
+                    "question": "单个座椅轴相对移动只能提供一种物理增量。",
+                }
+            )
+            return None
+        if vehicle_profile is None:
+            ambiguities.append(
+                {
+                    "field": f"requests[{index}].parameters.position_pct",
+                    "question": "缺少车型座椅量程，无法把相对物理移动换算为绝对百分比。",
+                }
+            )
+            return None
+
+        parameter_name = physical_parameters[0]
+        delta = parameters[parameter_name]
+        if (
+            not isinstance(delta, (int, float))
+            or isinstance(delta, bool)
+            or float(delta) <= 0
+        ):
+            ambiguities.append(
+                {
+                    "field": f"requests[{index}].parameters.{parameter_name}",
+                    "question": "座椅相对物理移动量必须是正数。",
+                }
+            )
+            return None
+
+        motor = vehicle_profile.seat_motors.get(axis)
+        if motor is None:
+            ambiguities.append(
+                {
+                    "field": f"requests[{index}].capability",
+                    "question": f"车型 {vehicle_profile.name} 未配置 {axis} 轴量程。",
+                }
+            )
+            return None
+        expected_unit = "mm" if parameter_name == "distance_mm" else "degree"
+        if motor.unit != expected_unit:
+            ambiguities.append(
+                {
+                    "field": f"requests[{index}].parameters.{parameter_name}",
+                    "question": (
+                        f"{axis} 轴使用 {motor.unit} 标定，不能用 {expected_unit} 增量换算。"
+                    ),
+                }
+            )
+            return None
+
+        direction = str(request.get("operation", "set"))
+        if direction not in {"increase", "decrease"}:
+            ambiguities.append(
+                {
+                    "field": f"requests[{index}].operation",
+                    "question": "相对物理移动需要 increase 或 decrease 方向。",
+                }
+            )
+            return None
+        previous = (
+            context.get("devices", {})
+            .get(target, {})
+            .get("position", {})
+            .get(axis)
+        )
+        if not isinstance(previous, dict):
+            ambiguities.append(
+                {
+                    "field": f"requests[{index}].target_scope",
+                    "question": "缺少当前座椅姿态，无法安全执行相对物理移动。",
+                }
+            )
+            return None
+        current_pct = previous.get("position_pct")
+        if (
+            not isinstance(current_pct, (int, float))
+            or isinstance(current_pct, bool)
+        ):
+            ambiguities.append(
+                {
+                    "field": f"requests[{index}].target_scope",
+                    "question": "当前座椅百分比位置无效，无法安全执行相对物理移动。",
+                }
+            )
+            return None
+
+        position_pct = vehicle_profile.relative_target_pct(
+            axis,
+            current_pct=current_pct,
+            delta=float(delta),
+            direction=direction,
+        )
+        action: dict[str, Any] = {
+            "action_type": "position",
+            "target": target,
+            "axis": axis,
+            "position_pct": position_pct,
+        }
+        if parameter_name == "angle_delta_deg":
+            action["angle_deg"] = ModelPlanAdapter._relative_target_angle(
+                previous,
+                motor,
+                position_pct,
+                float(delta),
+                direction,
+            )
+        return ModelPlanAdapter._preserve_relative_motion(
+            index, request, action, context, ambiguities
+        )
+
+    @staticmethod
+    def _relative_target_angle(
+        previous: dict[str, Any],
+        motor: Any,
+        position_pct: int,
+        delta_deg: float,
+        direction: str,
+    ) -> int | float:
+        start_angle = previous.get("angle_deg")
+        if (
+            isinstance(start_angle, (int, float))
+            and not isinstance(start_angle, bool)
+        ):
+            signed_delta = delta_deg if direction == "increase" else -delta_deg
+            return _clean_number(float(start_angle) + signed_delta)
+        display_angle = motor.display_angle_for_pct(position_pct)
+        if display_angle is None:
+            return _clean_number(delta_deg)
+        return _clean_number(display_angle)
+
     @staticmethod
     def _preserve_relative_motion(
         index: int,
@@ -530,3 +738,7 @@ class ModelPlanAdapter:
             )
         action["motion_intent"] = motion_intent
         return action
+
+
+def _clean_number(value: float) -> int | float:
+    return int(value) if float(value).is_integer() else round(float(value), 1)