File size: 2,929 Bytes
6835659 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | from typing import Any, Dict
from src.planner.schema import SemanticPlan
_LIST_FIELDS = {
"primary_entities",
"secondary_entities",
"visual_attributes",
"style",
"mood_emotion",
"narrative_tone",
"audio_intent",
"audio_elements",
"must_include",
"must_avoid",
}
_LIST_PATHS = [
("core_semantics", "main_subjects"),
("core_semantics", "actions"),
("style_controls", "visual_style"),
("style_controls", "color_palette"),
("style_controls", "lighting"),
("style_controls", "camera"),
("style_controls", "mood_emotion"),
("style_controls", "narrative_tone"),
("image_constraints", "must_include"),
("image_constraints", "must_avoid"),
("image_constraints", "objects"),
("image_constraints", "environment_details"),
("image_constraints", "composition"),
("audio_constraints", "audio_intent"),
("audio_constraints", "sound_sources"),
("audio_constraints", "ambience"),
("audio_constraints", "must_include"),
("audio_constraints", "must_avoid"),
("text_constraints", "must_include"),
("text_constraints", "must_avoid"),
("text_constraints", "keywords"),
]
_STRING_PATHS = [
("scene_summary",),
("domain",),
("core_semantics", "setting"),
("core_semantics", "time_of_day"),
("core_semantics", "weather"),
("audio_constraints", "tempo"),
("text_constraints", "length"),
]
def _as_list(value: Any) -> list[str]:
if value is None:
return []
if isinstance(value, list):
return [str(v) for v in value if str(v).strip()]
if isinstance(value, str):
return [value] if value.strip() else []
return [str(value)]
def _as_str(value: Any) -> str:
if value is None:
return ""
if isinstance(value, list):
for item in value:
text = str(item).strip()
if text:
return text
return ""
return str(value)
def _get_parent(data: Dict[str, Any], path: tuple[str, ...]) -> Dict[str, Any] | None:
cur: Any = data
for key in path[:-1]:
if not isinstance(cur, dict) or key not in cur:
return None
cur = cur[key]
if not isinstance(cur, dict):
return None
return cur
def _normalize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
for key in _LIST_FIELDS:
if key in data:
data[key] = _as_list(data[key])
for path in _LIST_PATHS:
parent = _get_parent(data, path)
if parent is not None and path[-1] in parent:
parent[path[-1]] = _as_list(parent[path[-1]])
for path in _STRING_PATHS:
parent = _get_parent(data, path)
if parent is not None and path[-1] in parent:
parent[path[-1]] = _as_str(parent[path[-1]])
return data
def validate_semantic_plan_dict(data: Dict[str, Any]) -> None:
data = _normalize_fields(data)
SemanticPlan(**data)
|