feat(video-routing): route real shot specs across direct/parallel/sequential

src/video_validation/__init__.py

@@ -0,0 +1,18 @@
+"""Video validation and generation-routing primitives for StoryCore."""
+
+from .multisubject_router import (
+    GenerationStrategy,
+    MultiSubjectShot,
+    RoutingDecision,
+    route_multi_subject_shot,
+)
+from .shot_spec_adapter import extract_multi_subject_shot, route_shot_spec
+
+__all__ = [
+    "GenerationStrategy",
+    "MultiSubjectShot",
+    "RoutingDecision",
+    "route_multi_subject_shot",
+    "extract_multi_subject_shot",
+    "route_shot_spec",
+]

src/video_validation/multisubject_router.py

@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+
+
+class GenerationStrategy(str, Enum):
+    DIRECT = "DIRECT"
+    PARALLEL = "PARALLEL"
+    SEQUENTIAL = "SEQUENTIAL"
+
+
+@dataclass(frozen=True)
+class MultiSubjectShot:
+    subject_count: int
+    interaction_strength: float = 0.0
+    contact_required: bool = False
+    occlusion_level: float = 0.0
+    identity_criticality: float = 0.5
+    camera_motion: float = 0.0
+    temporal_dependency: float = 0.0
+    compute_budget: float = 0.5
+
+    def __post_init__(self) -> None:
+        if self.subject_count < 0:
+            raise ValueError("subject_count must be non-negative")
+        for name in (
+            "interaction_strength",
+            "occlusion_level",
+            "identity_criticality",
+            "camera_motion",
+            "temporal_dependency",
+            "compute_budget",
+        ):
+            value = getattr(self, name)
+            if not 0.0 <= value <= 1.0:
+                raise ValueError(f"{name} must be in [0, 1]")
+
+
+@dataclass(frozen=True)
+class RoutingDecision:
+    strategy: GenerationStrategy
+    confidence: float
+    reasons: tuple[str, ...]
+    requires_intermediate_validation: bool
+
+
+def route_multi_subject_shot(shot: MultiSubjectShot) -> RoutingDecision:
+    """Choose a generation strategy using deterministic, explainable rules.
+
+    The router does not call a model and does not generate media. It only
+    chooses the safest/cheapest generation topology for the supplied shot
+    complexity. Downstream validators still decide whether the result passes.
+    """
+
+    reasons: list[str] = []
+
+    if shot.subject_count <= 1:
+        return RoutingDecision(
+            strategy=GenerationStrategy.DIRECT,
+            confidence=0.98,
+            reasons=("single_or_no_subject",),
+            requires_intermediate_validation=False,
+        )
+
+    # Strong physical/spatial coupling is the clearest case for staged
+    # generation because independent branches cannot reliably preserve contact.
+    if shot.contact_required:
+        reasons.append("contact_required")
+    if shot.interaction_strength >= 0.7:
+        reasons.append("strong_subject_interaction")
+    if shot.occlusion_level >= 0.75:
+        reasons.append("heavy_occlusion")
+    if shot.temporal_dependency >= 0.8:
+        reasons.append("strong_temporal_dependency")
+
+    if reasons:
+        confidence = min(0.98, 0.80 + 0.04 * len(reasons))
+        return RoutingDecision(
+            strategy=GenerationStrategy.SEQUENTIAL,
+            confidence=confidence,
+            reasons=tuple(reasons),
+            requires_intermediate_validation=True,
+        )
+
+    # Independent subjects with high identity requirements benefit from
+    # separate generation branches followed by composition/reconciliation.
+    parallel_score = 0.0
+    if shot.identity_criticality >= 0.7:
+        parallel_score += 0.45
+        reasons.append("identity_critical")
+    if shot.subject_count >= 3:
+        parallel_score += 0.25
+        reasons.append("many_subjects")
+    if shot.interaction_strength <= 0.35:
+        parallel_score += 0.20
+        reasons.append("weak_subject_interaction")
+    if shot.occlusion_level <= 0.35:
+        parallel_score += 0.10
+        reasons.append("low_occlusion")
+
+    if parallel_score >= 0.65:
+        return RoutingDecision(
+            strategy=GenerationStrategy.PARALLEL,
+            confidence=min(0.95, 0.70 + parallel_score * 0.25),
+            reasons=tuple(reasons),
+            requires_intermediate_validation=True,
+        )
+
+    # Complex camera motion and previous-shot continuity can still favor a
+    # staged build even without direct contact between subjects.
+    if shot.camera_motion >= 0.75 and shot.temporal_dependency >= 0.55:
+        return RoutingDecision(
+            strategy=GenerationStrategy.SEQUENTIAL,
+            confidence=0.78,
+            reasons=("camera_motion_with_temporal_dependency",),
+            requires_intermediate_validation=True,
+        )
+
+    # When compute is severely constrained, use the single-pass baseline and
+    # rely on validators rather than multiplying generation branches.
+    if shot.compute_budget <= 0.2:
+        return RoutingDecision(
+            strategy=GenerationStrategy.DIRECT,
+            confidence=0.72,
+            reasons=("compute_budget_constrained",),
+            requires_intermediate_validation=False,
+        )
+
+    return RoutingDecision(
+        strategy=GenerationStrategy.DIRECT,
+        confidence=0.70,
+        reasons=("low_interaction_complexity",),
+        requires_intermediate_validation=False,
+    )

src/video_validation/shot_spec_adapter.py

@@ -0,0 +1,163 @@
+from __future__ import annotations
+
+import math
+from collections.abc import Mapping, Sequence
+from typing import Any
+
+from .multisubject_router import MultiSubjectShot, RoutingDecision, route_multi_subject_shot
+
+
+_CAMERA_MOTION_INTENSITY = {
+    "static": 0.0,
+    "locked": 0.0,
+    "none": 0.0,
+    "pan": 0.35,
+    "tilt": 0.35,
+    "zoom": 0.45,
+    "dolly": 0.55,
+    "tracking": 0.6,
+    "truck": 0.6,
+    "pedestal": 0.55,
+    "orbit": 0.75,
+    "crane": 0.75,
+    "handheld": 0.8,
+    "whip": 0.9,
+}
+
+
+def _read(source: Any, key: str, default: Any = None) -> Any:
+    if isinstance(source, Mapping):
+        return source.get(key, default)
+    return getattr(source, key, default)
+
+
+def _routing_metadata(shot_spec: Any) -> Mapping[str, Any]:
+    metadata = _read(shot_spec, "metadata", {})
+    if not isinstance(metadata, Mapping):
+        return {}
+    routing = metadata.get("multi_subject_routing", {})
+    return routing if isinstance(routing, Mapping) else {}
+
+
+def _normalize_score(value: Any, *, name: str, default: float) -> float:
+    if value is None:
+        return default
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        raise ValueError(f"{name} must be numeric")
+    score = float(value)
+    if not math.isfinite(score) or not 0.0 <= score <= 1.0:
+        raise ValueError(f"{name} must be finite and in [0, 1]")
+    return score
+
+
+def _normalize_bool(value: Any, *, name: str, default: bool) -> bool:
+    if value is None:
+        return default
+    if not isinstance(value, bool):
+        raise ValueError(f"{name} must be boolean")
+    return value
+
+
+def _camera_motion_score(value: Any) -> float:
+    if value is None:
+        return 0.0
+    if isinstance(value, (int, float)) and not isinstance(value, bool):
+        return _normalize_score(value, name="camera_motion", default=0.0)
+    if not isinstance(value, str):
+        raise ValueError("camera_motion must be numeric or text")
+
+    normalized = value.lower().replace("-", " ").replace("_", " ")
+    matched = [score for token, score in _CAMERA_MOTION_INTENSITY.items() if token in normalized]
+    return max(matched, default=0.25 if normalized.strip() else 0.0)
+
+
+def _sequence_count(value: Any) -> int | None:
+    if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
+        return len(value)
+    return None
+
+
+def _subject_count(shot_spec: Any, *, subjects: Sequence[Any] | None, routing: Mapping[str, Any]) -> int:
+    candidates: list[tuple[str, int]] = []
+
+    explicit = routing.get("subject_count", _read(shot_spec, "subject_count", None))
+    if explicit is not None:
+        if isinstance(explicit, bool) or not isinstance(explicit, int) or explicit < 0:
+            raise ValueError("subject_count must be a non-negative integer")
+        candidates.append(("subject_count", explicit))
+
+    for key in ("characters_present", "subjects", "characters"):
+        count = _sequence_count(_read(shot_spec, key, None))
+        if count is not None:
+            candidates.append((key, count))
+
+    if subjects is not None:
+        candidates.append(("subjects_argument", len(subjects)))
+
+    if not candidates:
+        raise ValueError("subject_count is required when the shot spec has no subject list")
+
+    distinct = {count for _, count in candidates}
+    if len(distinct) != 1:
+        detail = ", ".join(f"{name}={count}" for name, count in candidates)
+        raise ValueError(f"contradictory subject counts: {detail}")
+    return candidates[0][1]
+
+
+def extract_multi_subject_shot(
+    shot_spec: Any,
+    *,
+    subjects: Sequence[Any] | None = None,
+    overrides: Mapping[str, Any] | None = None,
+) -> MultiSubjectShot:
+    """Adapt an existing StoryCore shot/dict into the deterministic routing contract.
+
+    Rich routing metadata can live under ``metadata.multi_subject_routing``.
+    ``overrides`` is intended for the orchestration layer when scene context knows
+    more than the shot object itself. No NLP guessing is performed here.
+    """
+
+    routing = dict(_routing_metadata(shot_spec))
+    if overrides:
+        routing.update(overrides)
+
+    camera_value = routing.get(
+        "camera_motion",
+        _read(shot_spec, "camera_movement", _read(shot_spec, "camera_motion", None)),
+    )
+
+    return MultiSubjectShot(
+        subject_count=_subject_count(shot_spec, subjects=subjects, routing=routing),
+        interaction_strength=_normalize_score(
+            routing.get("interaction_strength"), name="interaction_strength", default=0.0
+        ),
+        contact_required=_normalize_bool(
+            routing.get("contact_required"), name="contact_required", default=False
+        ),
+        occlusion_level=_normalize_score(
+            routing.get("occlusion_level"), name="occlusion_level", default=0.0
+        ),
+        identity_criticality=_normalize_score(
+            routing.get("identity_criticality"), name="identity_criticality", default=0.5
+        ),
+        camera_motion=_camera_motion_score(camera_value),
+        temporal_dependency=_normalize_score(
+            routing.get("temporal_dependency"), name="temporal_dependency", default=0.0
+        ),
+        compute_budget=_normalize_score(
+            routing.get("compute_budget"), name="compute_budget", default=0.5
+        ),
+    )
+
+
+def route_shot_spec(
+    shot_spec: Any,
+    *,
+    subjects: Sequence[Any] | None = None,
+    overrides: Mapping[str, Any] | None = None,
+) -> RoutingDecision:
+    """Extract a routing input from a StoryCore shot and choose its topology."""
+
+    return route_multi_subject_shot(
+        extract_multi_subject_shot(shot_spec, subjects=subjects, overrides=overrides)
+    )