Extract row camera policy

row_camera.py

@@ -0,0 +1,150 @@
+from __future__ import annotations
+
+from typing import Any, Callable, Mapping
+
+try:
+    from . import camera_config as camera_policy
+    from . import scene_camera_adapters
+except ImportError:  # Allows local smoke tests with top-level imports.
+    import camera_config as camera_policy
+    import scene_camera_adapters
+
+
+PovLabelResolver = Callable[[dict[str, Any]], list[str]]
+
+
+def _list_from(value: Any) -> list[Any]:
+    if value is None:
+        return []
+    if isinstance(value, list):
+        return value
+    return [value]
+
+
+def composition_prompt(composition: Any) -> str:
+    composition = str(composition or "").strip()
+    if not composition:
+        return composition
+    lower = composition.lower()
+    if lower.startswith("vertical ") or " vertical " in lower or lower.endswith(" vertical"):
+        return composition
+    return f"vertical {composition}"
+
+
+def insert_positive_directive(prompt: str, directive: str) -> str:
+    marker = " Avoid:"
+    if marker in prompt:
+        before, after = prompt.split(marker, 1)
+        return f"{before.rstrip()} {directive}{marker}{after}"
+    return f"{prompt.rstrip()} {directive}"
+
+
+def camera_caption_text(parsed: dict[str, Any]) -> str:
+    return camera_policy.camera_caption_text(parsed)
+
+
+def coworking_composition_prompt(scene_text: Any, composition: Any, subject_kind: str = "subjects") -> str:
+    return scene_camera_adapters.coworking_composition_prompt(scene_text, composition, subject_kind)
+
+
+def apply_contextual_composition(row: dict[str, Any], subject_kind: str) -> dict[str, Any]:
+    scene_text = row.get("scene_text") or row.get("source_scene_text") or row.get("scene")
+    old_composition = str(row.get("composition") or "").strip()
+    new_composition = coworking_composition_prompt(scene_text, old_composition, subject_kind)
+    if not old_composition or new_composition == old_composition:
+        return row
+    row["source_composition"] = row.get("source_composition") or old_composition
+    row["composition"] = new_composition
+    row["composition_prompt"] = composition_prompt(new_composition)
+    prompt = str(row.get("prompt") or "")
+    replacements = (
+        (f"Composition: vertical {old_composition}.", f"Composition: {composition_prompt(new_composition)}."),
+        (f"Composition: {old_composition}.", f"Composition: {composition_prompt(new_composition)}."),
+        (f"Framed as {old_composition}.", f"Framed as {new_composition}."),
+    )
+    for old_fragment, new_fragment in replacements:
+        if old_fragment in prompt:
+            row["prompt"] = prompt.replace(old_fragment, new_fragment)
+            break
+    row["caption"] = str(row.get("caption") or "").replace(f", {old_composition},", f", {new_composition},")
+    return row
+
+
+def camera_scene_directive_for_context(
+    scene_text: Any,
+    composition: Any,
+    camera_config: str | dict[str, Any] | None,
+    pov_labels: list[str] | None = None,
+    subject_kind: str = "subjects",
+    compact_labels: Mapping[str, str] | None = None,
+) -> tuple[str, dict[str, Any]]:
+    parsed = camera_policy.parse_camera_config(camera_config)
+    directive = scene_camera_adapters.camera_scene_directive_for_context(
+        scene_text,
+        parsed,
+        pov_labels,
+        subject_kind,
+        compact_labels,
+    )
+    return directive, parsed
+
+
+def row_camera_subject_kind(row: dict[str, Any]) -> str:
+    subject_type = str(row.get("subject_type") or row.get("primary_subject") or "").lower()
+    if subject_type in ("woman", "adult woman") or subject_type == "single_any":
+        return "woman"
+    if subject_type in ("man", "adult man"):
+        return "man"
+    try:
+        women_count = int(row.get("women_count") or 0)
+        men_count = int(row.get("men_count") or 0)
+    except (TypeError, ValueError):
+        women_count = men_count = 0
+    if women_count == 1 and men_count == 0:
+        return "woman"
+    if women_count == 0 and men_count == 1:
+        return "man"
+    if women_count + men_count == 2:
+        return "couple"
+    return "subjects"
+
+
+def row_pov_labels(row: dict[str, Any], resolver: PovLabelResolver | None = None) -> list[str]:
+    resolved: list[str] = []
+    if resolver is not None:
+        resolved = [str(label) for label in _list_from(resolver(row)) if str(label).strip()]
+    if resolved:
+        return resolved
+    return [str(label) for label in _list_from(row.get("pov_character_labels")) if str(label).strip()]
+
+
+def apply_camera_config(
+    row: dict[str, Any],
+    camera_config: str | dict[str, Any] | None,
+    *,
+    pov_label_resolver: PovLabelResolver | None = None,
+    compact_labels: Mapping[str, str] | None = None,
+) -> dict[str, Any]:
+    directive, parsed = camera_policy.camera_directive(camera_config)
+    pov_labels = row_pov_labels(row, pov_label_resolver)
+    subject_kind = row_camera_subject_kind(row)
+    row = apply_contextual_composition(row, subject_kind)
+    scene_directive, parsed = camera_scene_directive_for_context(
+        row.get("scene_text") or row.get("source_scene_text") or row.get("scene"),
+        row.get("composition") or row.get("source_composition"),
+        parsed,
+        pov_labels,
+        subject_kind,
+        compact_labels,
+    )
+    row["camera_config"] = parsed
+    row["camera_scene_directive"] = scene_directive
+    row["camera_directive"] = "" if pov_labels else directive
+    combined_directive = " ".join(part for part in (scene_directive, row["camera_directive"]) if part)
+    if not combined_directive:
+        return row
+    row["prompt"] = insert_positive_directive(str(row.get("prompt") or ""), combined_directive)
+    caption = camera_caption_text(parsed)
+    if caption and not pov_labels:
+        row["caption"] = f"{row.get('caption', '').rstrip()}, {caption}"
+    return row