139 lines
4.9 KiB
Python
139 lines
4.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
try:
|
|
from . import category_library as category_policy
|
|
from . import generate_prompt_batches as g
|
|
from . import location_config as location_policy
|
|
except ImportError: # Allows local smoke tests with top-level imports.
|
|
import category_library as category_policy
|
|
import generate_prompt_batches as g
|
|
import location_config as location_policy
|
|
|
|
|
|
def _list_from(value: Any) -> list[Any]:
|
|
if value is None:
|
|
return []
|
|
if isinstance(value, list):
|
|
return value
|
|
return [value]
|
|
|
|
|
|
def _is_false(value: Any) -> bool:
|
|
if isinstance(value, bool):
|
|
return value is False
|
|
if isinstance(value, str):
|
|
return value.strip().lower() in ("false", "0", "no", "off")
|
|
return False
|
|
|
|
|
|
def _unique_extend(target: list[Any], additions: list[Any]) -> None:
|
|
seen = set()
|
|
for item in target:
|
|
try:
|
|
seen.add(json.dumps(item, sort_keys=True))
|
|
except TypeError:
|
|
seen.add(repr(item))
|
|
for item in additions:
|
|
try:
|
|
marker = json.dumps(item, sort_keys=True)
|
|
except TypeError:
|
|
marker = repr(item)
|
|
if marker not in seen:
|
|
target.append(item)
|
|
seen.add(marker)
|
|
|
|
|
|
def scene_pool(
|
|
category: dict[str, Any],
|
|
subcategory: dict[str, Any],
|
|
item: Any,
|
|
subject_type: str,
|
|
location_config: dict[str, Any] | None = None,
|
|
) -> list[Any]:
|
|
location_config = location_config or {}
|
|
location_entries = _list_from(location_config.get("scene_entries"))
|
|
if location_policy.location_config_active(location_config) and location_config.get("apply_mode") == "replace":
|
|
return location_entries
|
|
fallback = g.GROUP_SCENES if subject_type in ("group", "configured_cast") else g.SCENES
|
|
scene_entries: list[Any] = []
|
|
scene_pools = category_policy.load_scene_pool_library()
|
|
item_source = item if isinstance(item, dict) else None
|
|
if item_source is not None and _is_false(item_source.get("inherit_scenes")):
|
|
sources = (item_source,)
|
|
elif _is_false(subcategory.get("inherit_scenes")):
|
|
sources = (subcategory, item_source)
|
|
else:
|
|
sources = (category, subcategory, item_source)
|
|
for source in sources:
|
|
if not isinstance(source, dict):
|
|
continue
|
|
if "scenes" in source:
|
|
_unique_extend(scene_entries, _list_from(source["scenes"]))
|
|
refs = _list_from(source.get("scene_pool")) + _list_from(source.get("scene_pools"))
|
|
for ref in refs:
|
|
ref_name = str(ref).strip()
|
|
if ref_name not in scene_pools:
|
|
raise ValueError(f"Unknown scene pool '{ref_name}'")
|
|
_unique_extend(scene_entries, scene_pools[ref_name])
|
|
if location_policy.location_config_active(location_config) and location_config.get("apply_mode") == "add":
|
|
_unique_extend(scene_entries, location_entries)
|
|
return scene_entries or fallback
|
|
|
|
|
|
def expression_pool(category: dict[str, Any], subcategory: dict[str, Any], item: Any) -> list[Any]:
|
|
return category_policy.configured_pool(
|
|
category,
|
|
subcategory,
|
|
item,
|
|
"expressions",
|
|
"expression_pools",
|
|
category_policy.load_expression_pool_library(),
|
|
"inherit_expressions",
|
|
) or g.EXPRESSIONS
|
|
|
|
|
|
def pose_pool(category: dict[str, Any], subcategory: dict[str, Any], item: Any, subject_type: str, poses: str) -> list[Any]:
|
|
configured = category_policy.merged_field(category, subcategory, item, "poses")
|
|
if configured:
|
|
return _list_from(configured)
|
|
if subject_type == "couple":
|
|
return [entry[2] for entry in g.COUPLE_TYPES]
|
|
if subject_type in ("layout", "scene"):
|
|
return ["clean designed layout"]
|
|
return g.EVOCATIVE_ALL if poses == "evocative" else g.POSES
|
|
|
|
|
|
def composition_pool(
|
|
category: dict[str, Any],
|
|
subcategory: dict[str, Any],
|
|
item: Any,
|
|
subject_type: str,
|
|
composition_config: dict[str, Any] | None = None,
|
|
) -> list[Any]:
|
|
composition_config = composition_config or {}
|
|
composition_entries = _list_from(composition_config.get("composition_entries"))
|
|
if location_policy.composition_config_active(composition_config) and composition_config.get("apply_mode") == "replace":
|
|
return composition_entries
|
|
configured = category_policy.configured_pool(
|
|
category,
|
|
subcategory,
|
|
item,
|
|
"compositions",
|
|
"composition_pools",
|
|
category_policy.load_composition_pool_library(),
|
|
"inherit_compositions",
|
|
)
|
|
if location_policy.composition_config_active(composition_config) and composition_config.get("apply_mode") == "add":
|
|
configured = list(configured or [])
|
|
_unique_extend(configured, composition_entries)
|
|
if configured:
|
|
return configured
|
|
if subject_type in ("group", "configured_cast"):
|
|
return g.GROUP_COMPOSITIONS
|
|
if subject_type in ("layout", "scene"):
|
|
return ["designed illustration layout"]
|
|
return g.COMPOSITIONS
|