feat: make output types workflow-first contracts

backend/app/domains/rendering/output_type_contracts.py

@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any, Literal
+
+from app.domains.rendering.workflow_config_utils import canonicalize_workflow_config
+from app.domains.rendering.workflow_node_registry import get_node_definition
+
+
+ResolvedWorkflowFamily = Literal["cad_file", "order_line", "mixed"]
+OutputTypeWorkflowFamily = Literal["cad_file", "order_line"]
+OutputTypeArtifactKind = Literal[
+    "still_image",
+    "turntable_video",
+    "model_export",
+    "thumbnail_image",
+    "blend_asset",
+    "package",
+    "custom",
+]
+
+_MODEL_EXPORT_FORMATS = {"gltf", "glb", "stl", "obj", "usd", "usdz"}
+_VIDEO_FORMATS = {"mp4", "webm", "mov"}
+_IMAGE_FORMATS = {"png", "jpg", "jpeg", "webp"}
+_ARTIFACT_KINDS_BY_FAMILY: dict[OutputTypeWorkflowFamily, set[OutputTypeArtifactKind]] = {
+    "cad_file": {"thumbnail_image", "model_export", "package", "custom"},
+    "order_line": {"still_image", "turntable_video", "blend_asset", "package", "custom"},
+}
+INVOCATION_OVERRIDE_KEYS = (
+    "width",
+    "height",
+    "engine",
+    "samples",
+    "frame_count",
+    "fps",
+    "turntable_axis",
+    "bg_color",
+    "noise_threshold",
+    "denoiser",
+    "denoising_input_passes",
+    "denoising_prefilter",
+    "denoising_quality",
+    "denoising_use_gpu",
+)
+
+
+def list_allowed_artifact_kinds_for_family(
+    workflow_family: str,
+) -> tuple[OutputTypeArtifactKind, ...]:
+    normalized_family = (workflow_family or "order_line").strip().lower()
+    if normalized_family == "cad_file":
+        allowed = _ARTIFACT_KINDS_BY_FAMILY["cad_file"]
+    else:
+        allowed = _ARTIFACT_KINDS_BY_FAMILY["order_line"]
+    return tuple(sorted(allowed))
+
+
+def infer_output_type_artifact_kind(
+    output_format: str | None,
+    is_animation: bool,
+    workflow_family: str = "order_line",
+) -> str:
+    normalized_format = (output_format or "").strip().lower()
+    normalized_family = (workflow_family or "order_line").strip().lower()
+
+    if is_animation or normalized_format in _VIDEO_FORMATS:
+        return "turntable_video"
+    if normalized_format in _MODEL_EXPORT_FORMATS:
+        return "model_export"
+    if normalized_family == "cad_file" and normalized_format in _IMAGE_FORMATS:
+        return "thumbnail_image"
+    return "still_image"
+
+
+def validate_output_type_contract(
+    *,
+    workflow_family: str,
+    artifact_kind: str,
+    output_format: str | None,
+    is_animation: bool,
+) -> None:
+    normalized_family = (workflow_family or "order_line").strip().lower()
+    normalized_artifact = (artifact_kind or "").strip().lower()
+    normalized_format = (output_format or "").strip().lower()
+
+    allowed_artifact_kinds = list_allowed_artifact_kinds_for_family(normalized_family)
+    if normalized_artifact not in allowed_artifact_kinds:
+        allowed = ", ".join(allowed_artifact_kinds)
+        raise ValueError(
+            f"Artifact kind '{artifact_kind}' is not allowed for workflow_family "
+            f"'{workflow_family}'. Allowed: {allowed}"
+        )
+
+    if normalized_family == "cad_file" and is_animation:
+        raise ValueError("CAD-file workflows do not support animated output types")
+
+    if normalized_artifact == "turntable_video":
+        if not is_animation:
+            raise ValueError("Artifact kind 'turntable_video' requires is_animation=true")
+        if normalized_format and normalized_format not in _VIDEO_FORMATS:
+            raise ValueError(
+                "Artifact kind 'turntable_video' requires a video output_format "
+                f"({', '.join(sorted(_VIDEO_FORMATS))})"
+            )
+
+    if normalized_artifact in {"still_image", "thumbnail_image"} and normalized_format in _VIDEO_FORMATS:
+        raise ValueError(
+            f"Artifact kind '{artifact_kind}' is incompatible with video output_format '{output_format}'"
+        )
+
+    if normalized_artifact == "model_export" and normalized_format and normalized_format not in _MODEL_EXPORT_FORMATS:
+        raise ValueError(
+            "Artifact kind 'model_export' requires a model output_format "
+            f"({', '.join(sorted(_MODEL_EXPORT_FORMATS))})"
+        )
+
+
+def infer_workflow_family_from_config(config: dict) -> ResolvedWorkflowFamily | None:
+    normalized = canonicalize_workflow_config(config)
+    families = {
+        definition.family
+        for node in normalized.get("nodes", [])
+        if (definition := get_node_definition(node.get("step"))) is not None
+    }
+    if not families:
+        return None
+    if len(families) > 1:
+        return "mixed"
+    return next(iter(families))
+
+
+def normalize_invocation_overrides(raw: Mapping[str, Any] | None) -> dict[str, Any]:
+    if not isinstance(raw, Mapping):
+        return {}
+    normalized: dict[str, Any] = {}
+    for key in INVOCATION_OVERRIDE_KEYS:
+        value = raw.get(key)
+        if value not in (None, ""):
+            normalized[key] = value
+    return normalized
+
+
+def merge_output_type_invocation_overrides(
+    render_settings: Mapping[str, Any] | None,
+    invocation_overrides: Mapping[str, Any] | None,
+) -> dict[str, Any]:
+    merged = normalize_invocation_overrides(render_settings)
+    merged.update(normalize_invocation_overrides(invocation_overrides))
+    return merged
+
+
+def apply_invocation_overrides_to_render_settings(
+    render_settings: Mapping[str, Any] | None,
+    invocation_overrides: Mapping[str, Any] | None,
+) -> dict[str, Any]:
+    merged = dict(render_settings or {})
+    normalized_invocation = normalize_invocation_overrides(invocation_overrides)
+    for key in INVOCATION_OVERRIDE_KEYS:
+        merged.pop(key, None)
+    merged.update(normalized_invocation)
+    return merged