"""Translate a generic model plan into domain adapter inputs without semantic rewriting.""" from __future__ import annotations from copy import deepcopy import re from typing import Any from cognitive_agent.application.native_can_intent import resolve_ch1_default_target from cognitive_agent.application.seat_calibration import ( ANGLE_CALIBRATIONS, angle_to_position_pct, ) from cognitive_agent.application.vehicle_profile import VehicleProfile SEAT_CAPABILITIES = { "seat.set_ventilation": "ventilation", "seat.set_heating": "heating", } SEAT_MOTION_CAPABILITIES = { "seat.set_longitudinal_position": "longitudinal", "seat.set_cushion_tilt": "cushion_tilt", "seat.set_backrest_recline": "backrest_recline", "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" DYNAMIC_MODE_CAPABILITY = "seat.start_dynamic_mode" DYNAMIC_MODE_STOP_CAPABILITY = "seat.stop_dynamic_mode" DYNAMIC_MODE_MAX_RUNTIME_MS = 600_000 TIMED_MOTION_MIN_DURATION_MS = 300 TIMED_AXIS_LIMITS = { "longitudinal": {"max_duration_ms": 30_000, "ms_per_pct": 120}, "cushion_tilt": {"max_duration_ms": 20_000, "ms_per_pct": 80}, "backrest_recline": {"max_duration_ms": 30_000, "ms_per_pct": 120}, "legrest_height": {"max_duration_ms": 25_000, "ms_per_pct": 100}, } HEX_COLOR = re.compile(r"^#[0-9A-Fa-f]{6}$") NATIVE_CAN_ONLY_CAPABILITIES = { "climate.set_control_mode", "roof.set_shade_motion", } DEFAULT_DRIVER_NATIVE_SEAT_CAPABILITIES = { "seat.set_ventilation": "seat.set_driver_ventilation_native", "seat.set_heating": "seat.set_driver_heating_native", } EXPLICIT_DRIVER_SEAT_PATTERN = re.compile( r"主驾|驾驶位|驾驶员|司机位|左前(?:的)?(?:座椅|座位)|" r"seat\s+(?:driver|front-left)", re.IGNORECASE, ) EXPLICIT_NON_DRIVER_SEAT_PATTERN = re.compile( r"副驾|乘客|乘员|右前|后排|后座|二排|第二排|中排|第三排|" r"前排|前舱|后舱|左侧|右侧|左边|右边|所有座椅|全部座椅|全车座椅|" r"seat\s+(?:passenger|front-right)", re.IGNORECASE, ) class ModelPlanAdapter: """Perform structural capability routing; user-intent decisions remain model-owned.""" def __init__(self, capability_context: dict[str, Any]) -> None: self.capabilities: dict[str, tuple[str, str, dict[str, Any]]] = {} for domain in capability_context.get("domains", []): for capability in domain.get("capabilities", []): self.capabilities[capability["name"]] = ( domain["domain"], domain["availability"], capability, ) def adapt( self, plan: dict[str, Any], context: dict[str, Any] | None = None, vehicle_profile: VehicleProfile | None = None, utterance: str = "", ) -> dict[str, Any]: actions: list[dict[str, Any]] = [] domain_actions: list[dict[str, Any]] = [] deferred: list[dict[str, Any]] = [] preset_actions: list[dict[str, Any]] = [] preset_events: list[dict[str, Any]] = [] ambiguities = list(plan.get("ambiguities", [])) context = context or {} for index, raw_request in enumerate(plan.get("requests", [])): request = self._apply_ch1_default_target(raw_request, utterance) capability_name = str(request.get("capability")) registered = self.capabilities.get(capability_name) if registered is None or registered[0] != request.get("domain"): ambiguities.append( { "field": f"requests[{index}].capability", "question": "模型请求的领域与能力注册表不一致。", } ) continue domain, domain_availability, capability = registered parameters = { str(item["name"]): item.get("value") for item in request.get("parameters", []) } 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( { "field": f"requests[{index}].parameters", "question": ( f"能力 {capability_name} 不支持参数 " f"{', '.join(sorted(unknown_parameters))}。" ), } ) continue availability = ( "native_can" if capability_name in NATIVE_CAN_ONLY_CAPABILITIES else domain_availability ) native_only_request = ( domain == "seat" and capability_name in SEAT_CAPABILITIES and ( request.get("target_scope") in {"left_rear", "right_rear"} or parameters.get("level") in {4, 5} ) ) or ( capability_name == "climate.set_outlet_mode" and ( request.get("target_scope") == "vehicle" or parameters.get("mode") not in {"face", "feet"} ) ) if availability != "executable" or native_only_request: values = [value for value in parameters.values() if value is not None] deferred.append( { "domain": domain, "capability": capability_name, "target_scope": request["target_scope"], "desired_change": request["operation"], "requested_value": values[0] if len(values) == 1 else None, "parameters": parameters, "reason": request["reason"], "execution_backend": ( "native_can" if availability == "native_can" or native_only_request else availability ), } ) continue if domain != "seat": generic_action = self._generic_action( index, request, parameters, capability, ambiguities, ) if generic_action is not None: domain_actions.append(generic_action) continue if capability_name == SAVE_PRESET_CAPABILITY: preset_action = self._preset_action( index, request, parameters, ambiguities ) if preset_action is not None: preset_actions.append(preset_action) continue if capability_name == APPLY_PRESET_CAPABILITY: preset_action = self._preset_action( index, request, parameters, ambiguities ) if preset_action is None: continue preset = self._find_preset( context.get("presets", []), preset_action["target"], preset_action["preset_name"], ) if preset is None: available = [ str(item.get("name")) for item in context.get("presets", []) if item.get("target") == preset_action["target"] ] suffix = ( f"当前可用预设：{', '.join(available)}。" if available else "当前座位尚未保存任何预设。" ) ambiguities.append( { "field": f"requests[{index}].parameters.preset_name", "question": ( f"未找到 {preset_action['target']} 的预设" f"“{preset_action['preset_name']}”。{suffix}" ), } ) continue for operation in preset["function_preset"]["spec"]["operations"]: operation_parameters = dict(operation["parameters"]) action = self._seat_action( index, { "capability": operation["capability"], "target_scope": preset_action["target"], "operation": "set", }, operation_parameters, ambiguities, context, vehicle_profile, ) if action is not None: actions.append(action) preset_events.append( { "action": "apply", "target": preset_action["target"], "preset_name": preset_action["preset_name"], } ) continue if capability_name == DYNAMIC_MODE_STOP_CAPABILITY: dynamic_action = self._dynamic_mode_stop_action( index, request, parameters, ambiguities ) if dynamic_action is not None: actions.append(dynamic_action) continue if capability_name == DYNAMIC_MODE_CAPABILITY: dynamic_action = self._dynamic_mode_action( index, request, parameters, ambiguities ) if dynamic_action is not None: actions.append(dynamic_action) continue action = self._seat_action( index, request, parameters, ambiguities, context, vehicle_profile ) if action is not None: actions.append(action) return { "detected_domains": list(plan.get("detected_domains", ["seat"])), "actions": actions, "domain_actions": domain_actions, "preset_actions": preset_actions, "preset_events": preset_events, "deferred_requests": deferred, "ambiguities": ambiguities, "decision_summary": list(plan.get("decision_summary", [])), "confidence": float(plan.get("confidence", 0)), "model_requests": deepcopy(plan.get("requests", [])), } @staticmethod def _apply_ch1_default_target( raw_request: dict[str, Any], utterance: str ) -> dict[str, Any]: """Normalize omitted model targets to driver-console CH1 defaults.""" request = deepcopy(raw_request) # Repository-backed mode and speech payloads already carry a validated # structured target. An empty utterance is not authorization to # reinterpret that persisted scope as a driver-console default. if not utterance.strip(): return request capability = str(request.get("capability", "")) current_target = str(request.get("target_scope") or "").strip().lower() if current_target not in {"", "unknown", "unspecified", "default"}: # The plan guard has already tied this explicit scope to the # server-owned utterance. Rewriting it afterwards is a TOCTOU # semantic change and can widen a local request to another zone. return request native_seat_capability = DEFAULT_DRIVER_NATIVE_SEAT_CAPABILITIES.get( capability ) if native_seat_capability is not None: explicit_non_driver = bool( EXPLICIT_NON_DRIVER_SEAT_PATTERN.search(utterance) ) explicit_driver = bool(EXPLICIT_DRIVER_SEAT_PATTERN.search(utterance)) implicit_driver = ( resolve_ch1_default_target(native_seat_capability, utterance) == "driver" ) if not explicit_non_driver and (explicit_driver or implicit_driver): request["capability"] = native_seat_capability request["target_scope"] = "driver" capability = native_seat_capability default_target = resolve_ch1_default_target(capability, utterance) if default_target is not None: request["target_scope"] = default_target return request @staticmethod def _generic_action( index: int, request: dict[str, Any], parameters: dict[str, Any], capability: dict[str, Any], ambiguities: list[dict[str, str]], ) -> dict[str, Any] | None: target = request.get("target_scope") if target not in capability.get("target_scopes", []): ambiguities.append( { "field": f"requests[{index}].target_scope", "question": ( f"{request['capability']} 不支持目标范围 {target!r}。" ), } ) return None parameter_specs = { item["name"]: item for item in capability.get("parameters", []) } state_spec = parameter_specs.get("state") if state_spec is not None and parameters.get("state") is None: inferred_state = { "activate": "on", "deactivate": "off", }.get(request.get("operation")) if inferred_state is not None: parameters["state"] = inferred_state for name, spec in parameter_specs.items(): value = parameters.get(name) if spec.get("required") and value is None: ambiguities.append( { "field": f"requests[{index}].parameters.{name}", "question": ( f"{request['capability']} 需要模型给出参数 {name} 的绝对目标值。" ), } ) return None if value is None: continue expected_type = spec.get("type") numeric = ( isinstance(value, (int, float)) and not isinstance(value, bool) ) type_valid = { "number": numeric, "integer": numeric and float(value).is_integer(), "string": isinstance(value, str), "boolean": isinstance(value, bool), "object": isinstance(value, dict), "array": isinstance(value, list), }.get(str(expected_type), False) if not type_valid: ambiguities.append( { "field": f"requests[{index}].parameters.{name}", "question": f"参数 {name} 的类型必须是 {expected_type}。", } ) return None if numeric and ( ("minimum" in spec and float(value) < float(spec["minimum"])) or ("maximum" in spec and float(value) > float(spec["maximum"])) ): ambiguities.append( { "field": f"requests[{index}].parameters.{name}", "question": ( f"参数 {name} 必须位于 " f"{spec.get('minimum', '-∞')} 到 {spec.get('maximum', '+∞')} 之间。" ), } ) return None if spec.get("allowed_values") and value not in spec["allowed_values"]: ambiguities.append( { "field": f"requests[{index}].parameters.{name}", "question": ( f"参数 {name} 必须是 " f"{', '.join(map(str, spec['allowed_values']))} 之一。" ), } ) return None if name == "color_hex" and not HEX_COLOR.fullmatch(str(value)): ambiguities.append( { "field": f"requests[{index}].parameters.color_hex", "question": "氛围灯颜色必须是 #RRGGBB 格式。", } ) return None normalized = {} for name, value in parameters.items(): spec = parameter_specs[name] normalized[name] = ( int(value) if spec.get("type") == "integer" else value.upper() if name == "color_hex" else value ) return { "domain": request["domain"], "capability": request["capability"], "target": str(target), "parameters": normalized, "operation": request.get("operation", "set"), "reason": request.get("reason", ""), } @staticmethod def _preset_action( index: int, request: dict[str, Any], parameters: dict[str, Any], ambiguities: list[dict[str, str]], ) -> dict[str, str] | None: target = request.get("target_scope") if target != "passenger": ambiguities.append( { "field": f"requests[{index}].target_scope", "question": "当前座椅能力仅控制副驾，请将目标指定为 passenger。", } ) return None preset_name = parameters.get("preset_name") if ( not isinstance(preset_name, str) or not preset_name.strip() or len(preset_name.strip()) > 64 ): ambiguities.append( { "field": f"requests[{index}].parameters.preset_name", "question": "保存或调用座椅姿态预设需要提供 1-64 个字符的名称。", } ) return None return {"target": target, "preset_name": preset_name.strip()} @staticmethod def _dynamic_mode_action( index: int, request: dict[str, Any], parameters: dict[str, Any], ambiguities: list[dict[str, str]], ) -> dict[str, Any] | None: target = request.get("target_scope") if target != "passenger": ambiguities.append( { "field": f"requests[{index}].target_scope", "question": "当前动态座椅模式仅控制副驾。", } ) return None mode_id = parameters.get("mode_id") if ( not isinstance(mode_id, (int, float)) or isinstance(mode_id, bool) or not float(mode_id).is_integer() or int(mode_id) not in {1, 2} ): ambiguities.append( { "field": f"requests[{index}].parameters.mode_id", "question": "动态座椅模式编号必须是 1（木马模式）或 2（动态摇篮）。", } ) return None normalized_mode_id = int(mode_id) return { "action_type": "dynamic_mode", "target": target, "mode_id": normalized_mode_id, "mode_name": ( "木马模式" if normalized_mode_id == 1 else "动态摇篮模式" ), "repeat": "until_stopped_or_timeout", "max_runtime_ms": DYNAMIC_MODE_MAX_RUNTIME_MS, } @staticmethod def _dynamic_mode_stop_action( index: int, request: dict[str, Any], parameters: dict[str, Any], ambiguities: list[dict[str, str]], ) -> dict[str, Any] | None: target = request.get("target_scope") if target != "passenger": ambiguities.append( { "field": f"requests[{index}].target_scope", "question": "当前动态座椅模式仅控制副驾。", } ) return None if parameters: ambiguities.append( { "field": f"requests[{index}].parameters", "question": "停止动态座椅模式不接受额外参数。", } ) return None return { "action_type": "dynamic_mode_stop", "target": target, } @staticmethod def _find_preset( presets: list[dict[str, Any]], target: str, preset_name: str, ) -> dict[str, Any] | None: expected = preset_name.casefold() return next( ( preset for preset in presets if preset.get("target") == target and str(preset.get("name", "")).casefold() == expected ), None, ) @staticmethod def _seat_action( index: int, request: dict[str, Any], 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 != "passenger": ambiguities.append( { "field": f"requests[{index}].target_scope", "question": "当前座椅能力仅控制副驾，请将目标指定为 passenger。", } ) return None capability = request["capability"] if capability in SEAT_MOTION_CAPABILITIES: axis = SEAT_MOTION_CAPABILITIES[capability] if request.get("duration_ms") is not None: return ModelPlanAdapter._timed_motion_action( index, request, ambiguities, context, target, axis, vehicle_profile, ) 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: if not isinstance(angle, (int, float)) or isinstance(angle, bool): ambiguities.append( { "field": f"requests[{index}].parameters.angle_deg", "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 exact_angle: int | float = ( int(angle) if float(angle).is_integer() else float(angle) ) action = { "action_type": "position", "target": target, "axis": axis, "position_pct": position_pct, "angle_deg": exact_angle, } return ModelPlanAdapter._preserve_relative_motion( index, request, action, context, ambiguities ) position = parameters.get("position_pct") if ( not isinstance(position, (int, float)) or isinstance(position, bool) or not float(position).is_integer() or not 0 <= float(position) <= 100 ): ambiguities.append( { "field": f"requests[{index}].parameters.position_pct", "question": "座椅位置动作需要模型给出 0-100 的绝对整数目标。", } ) return None action = { "action_type": "position", "target": target, "axis": axis, "position_pct": int(position), } return ModelPlanAdapter._preserve_relative_motion( index, request, action, context, ambiguities ) state = parameters.get("state") if state is None: state = {"activate": "on", "deactivate": "off"}.get(request.get("operation")) if state not in {"on", "off"}: ambiguities.append( { "field": f"requests[{index}].parameters.state", "question": "座椅动作需要明确的 on/off 目标状态。", } ) return None level = parameters.get("level") return { "target": target, "function": SEAT_CAPABILITIES[capability], "state": state, "requested_level": ( int(level) if isinstance(level, (int, float)) and not isinstance(level, bool) else None ), } @staticmethod def _timed_motion_action( index: int, request: dict[str, Any], ambiguities: list[dict[str, str]], context: dict[str, Any], target: str, axis: str, vehicle_profile: VehicleProfile | None, ) -> dict[str, Any] | None: duration_ms = request.get("duration_ms") if ( not isinstance(duration_ms, (int, float)) or isinstance(duration_ms, bool) or not float(duration_ms).is_integer() or float(duration_ms) < 0 ): ambiguities.append( { "field": f"requests[{index}].duration_ms", "question": "座椅按时长移动需要非负的整数毫秒数。", } ) 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 timing = TIMED_AXIS_LIMITS.get(axis) if timing is None: ambiguities.append( { "field": f"requests[{index}].capability", "question": "该座椅轴当前不支持按时长运动。", } ) 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) or not 0 <= float(current_pct) <= 100 ): ambiguities.append( { "field": f"requests[{index}].target_scope", "question": "当前座椅位置无效，无法安全执行按时长运动。", } ) return None requested_duration_ms = int(duration_ms) maximum_duration_ms = int(timing["max_duration_ms"]) applied_duration_ms = min( maximum_duration_ms, max(TIMED_MOTION_MIN_DURATION_MS, requested_duration_ms), ) estimated_delta_pct = applied_duration_ms / int(timing["ms_per_pct"]) signed_delta = estimated_delta_pct if direction == "increase" else -estimated_delta_pct raw_target = float(current_pct) + signed_delta target_pct = round(max(0.0, min(100.0, raw_target))) motion_intent: dict[str, Any] = { "mode": "timed", "direction": direction, "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": applied_duration_ms != requested_duration_ms, } if raw_target >= 100: motion_intent["endpoint_expected"] = "far" elif raw_target <= 0: motion_intent["endpoint_expected"] = "near" action: dict[str, Any] = { "action_type": "position", "target": target, "axis": axis, "position_pct": target_pct, "motion_intent": motion_intent, "user_notice": f"座椅轴将按时长执行{applied_duration_ms / 1000:g}秒。", } motor = ( vehicle_profile.seat_motors.get(axis) if vehicle_profile is not None else None ) if motor is not None and motor.display_angle is not None: angle = motor.display_angle_for_pct(target_pct) if angle is not None: action["angle_deg"] = _clean_number(float(angle)) return action @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, request: dict[str, Any], action: dict[str, Any], context: dict[str, Any], ambiguities: list[dict[str, str]], ) -> dict[str, Any] | None: """Keep the model's relative intent after it resolves an absolute target. The absolute target remains the public capability value. This context is also needed by the hardware compiler to choose the bounded timed/Hall path when the controller has not yet acquired an absolute endpoint calibration. """ requested_operation = str(request.get("operation", "set")) if requested_operation not in {"increase", "decrease"}: return action target = str(action["target"]) axis = str(action["axis"]) previous = ( context.get("devices", {}) .get(target, {}) .get("position", {}) .get(axis) ) if not isinstance(previous, dict): return action start_position = previous.get("position_pct") if not isinstance(start_position, (int, float)) or isinstance( start_position, bool ): return action target_position = int(action["position_pct"]) signed_delta = target_position - int(start_position) expected_sign = 1 if requested_operation == "increase" else -1 if signed_delta * expected_sign < 0: ambiguities.append( { "field": f"requests[{index}].operation", "question": ( "模型给出的相对移动方向与其解析出的绝对目标相互矛盾，" "无法安全生成电机方向。" ), } ) return None motion_intent: dict[str, Any] = { "mode": "relative", "direction": requested_operation, "start_position_pct": int(start_position), "delta_position_pct": abs(signed_delta), } start_angle = previous.get("angle_deg") target_angle = action.get("angle_deg") if ( isinstance(start_angle, (int, float)) and not isinstance(start_angle, bool) and isinstance(target_angle, (int, float)) and not isinstance(target_angle, bool) ): motion_intent["start_angle_deg"] = float(start_angle) motion_intent["delta_angle_deg"] = abs( float(target_angle) - float(start_angle) ) 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)