151 lines
5.6 KiB
Python
151 lines
5.6 KiB
Python
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
|