Cognitive_Agent/domains/seat/src/seat_domain/generator.py

"""Generate public intent and function configuration documents."""

from __future__ import annotations

from datetime import UTC, datetime
from typing import Any


TARGETS = {
    "driver": {"device": "seat", "zone": "front_left", "instance": "driver"},
    "passenger": {"device": "seat", "zone": "front_right", "instance": "passenger"},
}
CAPABILITIES = {
    "ventilation": "seat.set_ventilation",
    "heating": "seat.set_heating",
}
MOTION_CAPABILITIES = {
    "longitudinal": "seat.set_longitudinal_position",
    "cushion_tilt": "seat.set_cushion_tilt",
    "backrest_recline": "seat.set_backrest_recline",
    "legrest_height": "seat.set_legrest_height",
    "legrest_extension": "seat.set_legrest_extension",
}


def _now() -> str:
    return datetime.now(UTC).isoformat().replace("+00:00", "Z")


def build_intent_ir(
    parsed: dict[str, Any], utterance: str, request_id: str, provider: str
) -> dict[str, Any]:
    requests = []
    for action in parsed["actions"]:
        if action.get("action_type") == "position":
            parameters = {
                "axis": action["axis"],
                "position_pct": action["position_pct"],
            }
            if action.get("angle_deg") is not None:
                parameters["angle_deg"] = action["angle_deg"]
            requests.append(
                {
                    "domain": "seat",
                    "goal": (
                        f"Set {action['target']} {action['axis']} position "
                        f"to {action['position_pct']} percent"
                    ),
                    "target": TARGETS[action["target"]],
                    "parameters": parameters,
                    "source_span": utterance,
                }
            )
            continue
        parameters: dict[str, Any] = {
            "function": action["function"],
            "state": action["state"],
        }
        if action.get("requested_level") is not None:
            parameters["requested_level"] = action["requested_level"]
        requests.append(
            {
                "domain": "seat",
                "goal": f"Set {action['target']} {action['function']} {action['state']}",
                "target": TARGETS[action["target"]],
                "parameters": parameters,
                "source_span": utterance,
            }
        )

    for preset_action in parsed.get("preset_actions", []):
        requests.append(
            {
                "domain": "seat",
                "goal": (
                    f"Save the current {preset_action['target']} posture as "
                    f"{preset_action['preset_name']}"
                ),
                "target": TARGETS[preset_action["target"]],
                "parameters": {
                    "capability": "seat.save_posture_preset",
                    "preset_name": preset_action["preset_name"],
                },
                "source_span": utterance,
            }
        )

    for preset_event in parsed.get("preset_events", []):
        requests.append(
            {
                "domain": "seat",
                "goal": (
                    f"Apply saved posture {preset_event['preset_name']} to "
                    f"{preset_event['target']}"
                ),
                "target": TARGETS[preset_event["target"]],
                "parameters": {
                    "capability": "seat.apply_posture_preset",
                    "preset_name": preset_event["preset_name"],
                },
                "source_span": utterance,
            }
        )

    for deferred in parsed.get("deferred_requests", []):
        requests.append(
            {
                "domain": deferred["domain"],
                "goal": f"{deferred['desired_change']} {deferred['capability']}",
                "target": {
                    "device": deferred["domain"],
                    "zone": deferred["target_scope"],
                },
                "parameters": {
                    "capability": deferred["capability"],
                    "desired_change": deferred["desired_change"],
                    "requested_value": deferred.get("requested_value"),
                    "requested_parameters": deferred.get("parameters", {}),
                    "availability": "catalog_only",
                },
                "source_span": utterance,
            }
        )

    detected_domains = parsed.get("detected_domains", ["seat"])

    return {
        "apiVersion": "cognitive-agent/v1alpha1",
        "kind": "IntentIR",
        "metadata": {
            "id": f"intent-{request_id}",
            "name": "Seat function intent",
            "version": "1.0.0",
            "request_id": request_id,
            "generated_at": _now(),
            "labels": {"provider": provider},
        },
        "spec": {
            "utterance": utterance,
            "language": "zh-CN",
            "mode": "cross_domain" if len(detected_domains) > 1 else "single_domain",
            "detected_domains": detected_domains,
            "requests": requests,
            "ambiguities": parsed["ambiguities"],
            "confidence": parsed["confidence"],
            "requires_confirmation": bool(parsed["ambiguities"]),
        },
    }


def build_function_config(
    parsed: dict[str, Any], request_id: str, provider: str
) -> dict[str, Any] | None:
    if not parsed["actions"]:
        return None

    operations = []
    assumptions = [
        "The supplied matrix is partial; this configuration is eligible for virtual execution only."
    ]
    for index, action in enumerate(parsed["actions"], start=1):
        if action.get("action_type") == "position":
            operation_id = (
                f"{action['target']}_{action['axis']}_{action['position_pct']}pct"
            )
            if any(item["id"] == operation_id for item in operations):
                operation_id = f"{operation_id}_{index}"
            parameters: dict[str, Any] = {"position_pct": action["position_pct"]}
            if action.get("angle_deg") is not None:
                parameters["angle_deg"] = action["angle_deg"]
            operation = {
                "id": operation_id,
                "capability": MOTION_CAPABILITIES[action["axis"]],
                "target": TARGETS[action["target"]],
                "parameters": parameters,
                "timeout_ms": 8000,
                "failure_policy": "abort",
            }
            if action.get("motion_intent") is not None:
                operation["motion"] = dict(action["motion_intent"])
            operations.append(operation)
            assumptions.append(
                f"Operation {operation_id} uses an explicitly synthetic position CAN mapping; "
                "it is never eligible for hardware transmission."
            )
            continue
        operation_id = f"{action['target']}_{action['function']}_{action['state']}"
        if any(item["id"] == operation_id for item in operations):
            operation_id = f"{operation_id}_{index}"
        parameters: dict[str, Any] = {"state": action["state"]}
        if action.get("requested_level") is not None:
            parameters["level"] = action["requested_level"]
        operations.append(
            {
                "id": operation_id,
                "capability": CAPABILITIES[action["function"]],
                "target": TARGETS[action["target"]],
                "parameters": parameters,
                "timeout_ms": 3000,
                "failure_policy": "abort",
            }
        )
        if action.get("requested_level") is not None:
            assumptions.append(
                f"Operation {operation_id} targets level {action['requested_level']}; "
                "the level is simulated because the supplied matrix has no level command signal."
            )

    return {
        "apiVersion": "cognitive-agent/v1alpha1",
        "kind": "FunctionConfig",
        "metadata": {
            "id": f"seat-{request_id}",
            "name": "Seat function",
            "version": "1.0.0",
            "request_id": request_id,
            "generated_at": _now(),
            "labels": {"provider": provider, "execution_scope": "virtual_only"},
        },
        "spec": {"domain": "seat", "operations": operations, "assumptions": assumptions},
    }


def build_function_preset(
    target: str,
    preset_name: str,
    device_state: dict[str, Any],
    request_id: str,
    provider: str,
) -> dict[str, Any]:
    operations = []
    for axis, capability in MOTION_CAPABILITIES.items():
        motion_state = device_state["position"][axis]
        parameters: dict[str, Any] = {
            "position_pct": int(motion_state["position_pct"])
        }
        if motion_state.get("angle_deg") is not None:
            parameters["angle_deg"] = motion_state["angle_deg"]
        operations.append(
            {
                "id": f"{target}_{axis}_preset",
                "capability": capability,
                "target": TARGETS[target],
                "parameters": parameters,
                "timeout_ms": 8000,
                "failure_policy": "abort",
            }
        )
    return {
        "apiVersion": "cognitive-agent/v1alpha1",
        "kind": "FunctionPreset",
        "metadata": {
            "id": f"seat-preset-{request_id}",
            "name": preset_name,
            "version": "1.0.0",
            "request_id": request_id,
            "generated_at": _now(),
            "description": f"User-saved five-axis posture for {target}.",
            "labels": {
                "provider": provider,
                "target": target,
                "storage_scope": "conversation_durable",
            },
        },
        "spec": {
            "domain": "seat",
            "operations": operations,
            "locked_parameters": [
                f"operations.{index}.parameters.position_pct"
                for index in range(len(operations))
            ],
            "applicability": {
                "target": target,
                "execution_mode": "dry_run",
            },
        },
    }