| |
| """ |
| Audio-lipsync card renderer — vertical PIL card scoped to the |
| ``docs/automations/LTX-2.3/audio-lipsync/`` iteration loop. |
| |
| Sibling to ``scripts/make_card.py`` (base-1step). Shares palette, |
| font fallback chain, and section primitives so the two families read |
| as a set, but the data sources are deliberately narrower: |
| |
| Two — and only two — source families feed this card: |
| |
| 1. The five ``Text Multiline`` nodes tracked by the loop config |
| (name / relay_overrides / overlay - info / overlay - feedback / |
| working_folder). |
| 2. The active ``LTXDirector`` node's own widget values and |
| input-socket wiring (Koolook for modified runs, upstream original |
| for A/B comparison; epsilon, frame_rate, timeline_data segments + |
| audioSegments, use_custom_audio toggle, audio_vae link state). |
| |
| Notably absent: BasicScheduler / KSamplerSelect / RandomNoise / |
| CFGGuider widget scrapes, ``_dev_build.json`` fork-state, ``git status`` |
| output. Those don't define what this loop sweeps. Adding them would |
| push the card into showing values the maintainer's curated multiline |
| notes don't claim to summarise. |
| |
| Card sections (top to bottom): |
| |
| HEADER run-NNN — {name} · date · job · workflow filename |
| KNOB STATE relay_overrides (the per-render knob) |
| BASE / RUN source workflow · copied-from workflow · working folder |
| DIRECTOR READINGS Director · pin tag · epsilon · Audio src |
| SEGMENTS segment list with time ranges + coverage rows |
| POST-RENDER feedback body + outcome scores |
| |
| Two entry points: |
| |
| 1. From loop_audio.py at end-of-snapshot: |
| |
| from scripts.make_card_audio import render_audio_card |
| render_audio_card(state, run_dir / "card.png") |
| |
| 2. Standalone, against an existing run folder (re-render after a |
| layout tweak): |
| |
| python scripts/make_card_audio.py <run_dir> |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import re |
| import sys |
| from pathlib import Path |
| from typing import Any |
|
|
| from PIL import Image, ImageDraw, ImageFont |
| from PIL.PngImagePlugin import PngInfo |
|
|
|
|
| |
| |
| |
|
|
| W = 540 |
| PAD_X = 28 |
| PAD_TOP = 28 |
| PAD_BOTTOM = 28 |
| BG_OUTER = (14, 14, 14) |
| BG_CARD = (21, 21, 21) |
| BG_SECTION = (26, 26, 31) |
| BORDER = (48, 47, 47) |
| TEXT = (249, 250, 251) |
| MUTED = (143, 149, 156) |
| DIM = (200, 204, 209) |
| ACCENT_RUN = (255, 184, 77) |
| ACCENT_BASE = (109, 180, 255) |
| ACCENT_OUT = (123, 207, 128) |
| NOTE_BG = (12, 12, 12) |
| RADIUS = 14 |
| SECTION_RADIUS = 8 |
|
|
|
|
| |
| |
| |
|
|
| WIN_FONTS = Path(r"C:/Windows/Fonts") |
|
|
|
|
| def _load_font(filenames: list[str], size: int) -> ImageFont.FreeTypeFont: |
| for fn in filenames: |
| for p in (WIN_FONTS / fn, Path("/Library/Fonts") / fn, |
| Path("/usr/share/fonts") / fn): |
| if p.exists(): |
| try: |
| return ImageFont.truetype(str(p), size) |
| except OSError: |
| continue |
| return ImageFont.load_default() |
|
|
|
|
| F_TITLE = _load_font(["segoeuib.ttf", "Arial Bold.ttf"], 28) |
| F_SUB = _load_font(["segoeui.ttf", "Arial.ttf"], 17) |
| F_H2 = _load_font(["segoeuib.ttf", "Arial Bold.ttf"], 16) |
| F_TAG = _load_font(["segoeuib.ttf", "Arial Bold.ttf"], 12) |
| F_SECTION = _load_font(["segoeuib.ttf", "Arial Bold.ttf"], 17) |
| F_MONO = _load_font(["consola.ttf", "Menlo.ttc", "DejaVuSansMono.ttf"], 19) |
| F_NOTE = _load_font(["segoeuii.ttf", "Arial Italic.ttf"], 18) |
|
|
|
|
| |
| |
| |
|
|
|
|
| def _wrap_text(text: str, max_chars: int, keep_blank_lines: bool = False) -> list[str]: |
| out: list[str] = [] |
| for raw_line in text.split("\n"): |
| if not raw_line.strip(): |
| if keep_blank_lines: |
| out.append("") |
| continue |
| words = raw_line.split() |
| cur = "" |
| for w in words: |
| if len(cur) + len(w) + 1 <= max_chars: |
| cur = (cur + " " + w).strip() |
| else: |
| if cur: |
| out.append(cur) |
| cur = w |
| if cur: |
| out.append(cur) |
| return out |
|
|
|
|
| def _draw_kv_row( |
| draw: ImageDraw.ImageDraw, x: int, y: int, |
| key: str, val: str, key_w: int, max_w: int | None = None, |
| ) -> int: |
| draw.text((x, y), key, font=F_MONO, fill=MUTED) |
| value = str(val) |
| if max_w is not None: |
| value = _trim_middle_to_width(draw, value, F_MONO, max_w) |
| draw.text((x + key_w, y), value, font=F_MONO, fill=TEXT) |
| return y + 26 |
|
|
|
|
| def _trim_to_width( |
| draw: ImageDraw.ImageDraw, |
| text: str, |
| font: ImageFont.FreeTypeFont, |
| max_w: int, |
| ) -> str: |
| if draw.textbbox((0, 0), text, font=font)[2] <= max_w: |
| return text |
| ellipsis = "..." |
| lo = 0 |
| hi = len(text) |
| while lo < hi: |
| mid = (lo + hi + 1) // 2 |
| candidate = text[:mid].rstrip() + ellipsis |
| if draw.textbbox((0, 0), candidate, font=font)[2] <= max_w: |
| lo = mid |
| else: |
| hi = mid - 1 |
| return text[:lo].rstrip() + ellipsis |
|
|
|
|
| def _trim_middle_to_width( |
| draw: ImageDraw.ImageDraw, |
| text: str, |
| font: ImageFont.FreeTypeFont, |
| max_w: int, |
| ) -> str: |
| if draw.textbbox((0, 0), text, font=font)[2] <= max_w: |
| return text |
| ellipsis = "..." |
| keep_tail = min(18, max(8, len(text) // 3)) |
| tail = text[-keep_tail:] |
| lo = 0 |
| hi = max(0, len(text) - keep_tail) |
| while lo < hi: |
| mid = (lo + hi + 1) // 2 |
| candidate = text[:mid].rstrip("_- ") + ellipsis + tail |
| if draw.textbbox((0, 0), candidate, font=font)[2] <= max_w: |
| lo = mid |
| else: |
| hi = mid - 1 |
| return text[:lo].rstrip("_- ") + ellipsis + tail |
|
|
|
|
| def _draw_text_box( |
| draw: ImageDraw.ImageDraw, x: int, y: int, width: int, |
| label: str, content: str, accent: tuple[int, int, int], |
| max_lines: int = 4, char_per_line: int = 42, |
| keep_blank_lines: bool = False, |
| ) -> int: |
| """Left-accented note block — colored 4-px bar on the left, label |
| in the accent colour, italic body text.""" |
| lines = _wrap_text(content, char_per_line, keep_blank_lines=keep_blank_lines)[:max_lines] |
| box_h = 28 + 22 * max(1, len(lines)) + 14 |
| draw.rounded_rectangle( |
| [x, y, x + width, y + box_h], |
| radius=SECTION_RADIUS, fill=BG_SECTION, outline=BORDER, width=1, |
| ) |
| draw.text((x + 12, y + 8), label.upper(), font=F_H2, fill=accent) |
| line_y = y + 36 |
| for line in lines: |
| draw.text((x + 12, line_y), line, font=F_NOTE, fill=DIM) |
| line_y += 22 |
| return y + box_h + 10 |
|
|
|
|
| def _draw_header_box( |
| draw: ImageDraw.ImageDraw, |
| x: int, |
| y: int, |
| width: int, |
| run_line: str, |
| module_name: str, |
| name_line: str, |
| ) -> int: |
| box_h = 118 |
| draw.rounded_rectangle( |
| [x, y, x + width, y + box_h], |
| radius=SECTION_RADIUS, fill=BG_SECTION, outline=BORDER, width=1, |
| ) |
| tx = x + 14 |
| ty = y + 14 |
| run_id, _, run_date = run_line.partition(" - ") |
| draw.text((tx, ty), run_id, font=F_H2, fill=ACCENT_RUN) |
| run_w = draw.textbbox((0, 0), run_id + " ", font=F_H2)[2] |
| if run_date: |
| draw.text((tx + run_w, ty), run_date, font=F_SUB, fill=MUTED) |
| ty += 27 |
| module_label = "Module:" |
| draw.text((tx, ty), module_label, font=F_SUB, fill=ACCENT_BASE) |
| label_w = draw.textbbox((0, 0), module_label + " ", font=F_SUB)[2] |
| draw.text((tx + label_w, ty), module_name, font=F_SUB, fill=TEXT) |
| ty += 31 |
| draw.text((tx, ty), name_line, font=F_TITLE, fill=TEXT) |
| return y + box_h + 10 |
|
|
|
|
| def _draw_section( |
| draw: ImageDraw.ImageDraw, x: int, y: int, width: int, |
| accent: tuple[int, int, int], label: str, body_h: int, |
| bg: tuple[int, int, int] = BG_SECTION, |
| border: tuple[int, int, int] = BORDER, |
| ) -> tuple[int, int, int, int]: |
| """Subtle panel with an accent-coloured uppercase label at the top. |
| Returns (content_x, content_y, content_w, end_y) so the caller |
| stacks rows directly underneath.""" |
| HEADER_H = 36 |
| section_h = HEADER_H + body_h + 14 |
| draw.rounded_rectangle( |
| [x, y, x + width, y + section_h], |
| radius=SECTION_RADIUS, fill=bg, outline=border, width=1, |
| ) |
| draw.text((x + 14, y + 10), label.upper(), font=F_SECTION, fill=accent) |
| return x + 14, y + HEADER_H, width - 28, y + section_h + 10 |
|
|
|
|
| def _section_body_rows(num_rows: int, extra: int = 0) -> int: |
| return num_rows * 26 + extra |
|
|
|
|
| def _draw_score_chip( |
| draw: ImageDraw.ImageDraw, |
| x: int, |
| y: int, |
| width: int, |
| label: str, |
| value: Any, |
| ) -> None: |
| draw.rounded_rectangle( |
| [x, y, x + width, y + 50], |
| radius=6, fill=BG_CARD, outline=BORDER, width=1, |
| ) |
| label_w = draw.textbbox((0, 0), label, font=F_TAG)[2] |
| draw.text((x + (width - label_w) // 2, y + 7), label, font=F_TAG, fill=ACCENT_OUT) |
| score = f"{value}/5" if value is not None else "?/5" |
| score_w = draw.textbbox((0, 0), score, font=F_MONO)[2] |
| draw.text((x + (width - score_w) // 2, y + 24), score, font=F_MONO, fill=TEXT) |
|
|
|
|
| |
| |
| |
| |
| |
|
|
|
|
| |
| |
| |
| |
| sys.path.insert(0, str(Path(__file__).resolve().parent)) |
| from loop_audio import ( |
| copy_delivery_card, |
| video_segment_has_audio, |
| metadata_numeric_widget, |
| parse_dev_sync_audio_line, |
| repo_metadata_from_build, |
| render_notes_md, |
| scrub_path_for_metadata, |
| wrap_path, |
| ) |
|
|
|
|
| def _audio_source_label(audio_src: str) -> str: |
| labels = { |
| "custom": "custom audio ON", |
| "custom (empty)": "custom audio ON (empty)", |
| "model-gen": "model audio", |
| "off (no VAE)": "audio OFF", |
| "(no director)": "no director", |
| } |
| return labels.get(audio_src, audio_src) |
|
|
|
|
| def _read_existing_metadata(run_dir: Path) -> dict[str, Any]: |
| path = run_dir / "metadata.json" |
| if not path.is_file(): |
| return {} |
| try: |
| data = json.loads(path.read_text(encoding="utf-8")) |
| except (OSError, json.JSONDecodeError): |
| return {} |
| return data if isinstance(data, dict) else {} |
|
|
|
|
| def _preserved_run_date(run_dir: Path, existing_metadata: dict[str, Any]) -> str: |
| existing_date = (existing_metadata.get("run") or {}).get("date", "") |
| if isinstance(existing_date, str) and existing_date.strip(): |
| return existing_date.strip() |
| wf = next(run_dir.glob("run*_workflow.json"), None) |
| if wf is not None: |
| try: |
| from datetime import datetime |
| return datetime.fromtimestamp(wf.stat().st_mtime).date().isoformat() |
| except OSError: |
| pass |
| from datetime import date |
| return date.today().isoformat() |
|
|
|
|
| def render_audio_card(state: dict[str, Any], out_path: Path) -> Path: |
| """Render the audio-lipsync card PNG. ``state`` keys consumed: |
| |
| run_number, run_label, date, workflow_name |
| name, relay_overrides_raw, info_body, feedback_lines, scores, |
| work_folder, output_folder, output_name |
| director_variant, director_flavor, audio_src, epsilon, frame_rate, |
| segments, audio_segments, segment_prompt_mode |
| """ |
| name = state.get("name") or "(unnamed)" |
| relay = (state.get("relay_overrides_raw") or "").strip() |
| info_body = (state.get("info_body") or "").rstrip() |
| feedback_lines = state.get("feedback_lines") or [] |
| scores = state.get("scores") or {} |
| work_folder = state.get("work_folder") or "" |
| output_folder = state.get("output_folder") or "" |
| output_name = state.get("output_name") or "" |
| metadata = state.get("metadata") or {} |
| director_variant = state.get("director_variant") or "(missing)" |
| director_flavor = state.get("director_flavor") or director_variant |
| director_pin_tag = state.get("director_pin_tag") or ( |
| (metadata.get("director") or {}).get("pin_tag", "") |
| ) |
| audio_src = state.get("audio_src") or "?" |
| epsilon = state.get("epsilon") |
| fps = state.get("frame_rate") |
| segments = state.get("segments") or [] |
| audio_segs = state.get("audio_segments") or [] |
| prompt_mode = state.get("segment_prompt_mode") or "none" |
| run_meta = metadata.get("run") or {} |
| |
| |
| |
|
|
| canvas_h = 2400 |
| img = Image.new("RGB", (W, canvas_h), BG_OUTER) |
| draw = ImageDraw.Draw(img) |
| inset = 18 |
| draw.rounded_rectangle( |
| [inset, inset, W - inset, canvas_h - inset], |
| radius=RADIUS, fill=BG_CARD, outline=BORDER, width=1, |
| ) |
| x = inset + PAD_X |
| y = inset + PAD_TOP |
| inner_w = W - 2 * inset - 2 * PAD_X |
| |
| |
| key_w = 178 |
|
|
| |
| module_name = state.get("module_name") or "audio-lipsync" |
| run_line = f"Run {state['run_number']:03d} - {state['date']}" |
| name_line = _trim_to_width(draw, str(name), F_TITLE, inner_w) |
| y = _draw_header_box(draw, x, y, inner_w, run_line, module_name, name_line) |
|
|
| |
| path_lines = wrap_path(work_folder, max_chars=40) if work_folder else [] |
| output_lines = wrap_path(output_folder, max_chars=40) if output_folder else [] |
| body_h = 0 |
| if output_name: |
| body_h += 52 |
| source_workflow = state.get("source_workflow_name") or run_meta.get("workflow") |
| if source_workflow: |
| source_workflow = str(Path(str(source_workflow)).with_suffix("")) |
| source_workflow_lines = ( |
| wrap_path(str(source_workflow), max_chars=40) if source_workflow else [] |
| ) |
| if source_workflow_lines: |
| body_h += 26 + 26 * len(source_workflow_lines) + 4 |
| if path_lines: |
| body_h += 26 + 26 * len(path_lines) + 4 |
| if output_lines: |
| body_h += 26 + 26 * len(output_lines) + 4 |
| cx, cy, cw, end_y = _draw_section( |
| draw, x, y, inner_w, ACCENT_BASE, "Base / run", body_h, |
| ) |
| if source_workflow_lines: |
| draw.text((cx, cy), "Copied from", font=F_MONO, fill=MUTED) |
| cy += 26 |
| for line in source_workflow_lines: |
| draw.text((cx + 12, cy), line, font=F_MONO, fill=TEXT) |
| cy += 26 |
| cy += 4 |
| if output_name: |
| draw.text((cx, cy), "Output name", font=F_MONO, fill=MUTED) |
| cy += 26 |
| draw.text( |
| (cx + 12, cy), |
| _trim_middle_to_width(draw, output_name, F_MONO, cw - 12), |
| font=F_MONO, |
| fill=TEXT, |
| ) |
| cy += 26 |
| if path_lines: |
| draw.text((cx, cy), "Working folder", font=F_MONO, fill=MUTED) |
| cy += 26 |
| for line in path_lines: |
| draw.text((cx + 12, cy), line, font=F_MONO, fill=TEXT) |
| cy += 26 |
| if output_lines: |
| draw.text((cx, cy), "Output folder", font=F_MONO, fill=MUTED) |
| cy += 26 |
| for line in output_lines: |
| draw.text((cx + 12, cy), line, font=F_MONO, fill=TEXT) |
| cy += 26 |
| y = end_y |
|
|
| |
| seg_rows = min(len(segments), 6) |
| indent_seg = 18 |
| director_rows = 5 if epsilon is not None else 4 |
| body_h = 26 * director_rows + 26 * 3 + 26 * seg_rows + 8 + 26 * 3 + 4 |
| cx, cy, cw, end_y = _draw_section( |
| draw, x, y, inner_w, ACCENT_BASE, "Director readings", body_h, |
| ) |
| cy = _draw_kv_row( |
| draw, cx, cy, "Director", str(director_flavor), |
| key_w, cw - key_w, |
| ) |
| cy = _draw_kv_row( |
| draw, cx, cy, "Pin tag", |
| _trim_middle_to_width(draw, str(director_pin_tag), F_MONO, cw - key_w), |
| key_w, cw - key_w, |
| ) |
| if epsilon is not None: |
| cy = _draw_kv_row(draw, cx, cy, "epsilon", str(epsilon), key_w) |
| cy = _draw_kv_row(draw, cx, cy, "Audio", _audio_source_label(audio_src), key_w) |
| cy = _draw_kv_row(draw, cx, cy, "Frame rate", str(fps or "?"), key_w) |
| cy = _draw_kv_row(draw, cx, cy, "Image segments", f"({len(segments)})", key_w) |
| cy = _draw_kv_row(draw, cx, cy, "Audio segments", f"({len(audio_segs)})", key_w) |
| cy = _draw_kv_row(draw, cx, cy, "Prompt mode", prompt_mode, key_w) |
|
|
| visible = segments[:seg_rows] |
| all_have_prompt = bool(visible) and all((s.get("prompt") or "") for s in visible) |
| all_have_audio = bool(visible) and all( |
| video_segment_has_audio(s, audio_segs) for s in visible |
| ) |
| all_have_keyframe = bool(visible) and all(s.get("imageFile") for s in visible) |
|
|
| for i, seg in enumerate(visible): |
| start = seg.get("start", 0) |
| length = seg.get("length", 0) |
| prompt = seg.get("prompt") or "" |
| has_p = bool(prompt) |
| has_a = video_segment_has_audio(seg, audio_segs) |
| has_k = bool(seg.get("imageFile")) |
| header_col = ACCENT_OUT if (has_p and has_a and has_k) else ACCENT_BASE |
| if fps: |
| start_s = start / fps |
| end_s = (start + length) / fps |
| header = f"{i+1}) {start_s:.0f} to {end_s:.0f} seconds" |
| else: |
| header = f"{i+1}) frames {start}-{start + length} (no fps)" |
| draw.text((cx + indent_seg, cy), header, font=F_MONO, fill=header_col) |
| cy += 26 |
| cy += 8 |
|
|
| for label, present in ( |
| ("Prompt", all_have_prompt), |
| ("Audio", all_have_audio), |
| ("Keyframe", all_have_keyframe), |
| ): |
| mark = "[x]" if present else "[ ]" |
| mark_col = ACCENT_OUT if present else MUTED |
| draw.text((cx, cy), label, font=F_MONO, fill=MUTED) |
| draw.text((cx + key_w, cy), mark, font=F_MONO, fill=mark_col) |
| cy += 26 |
| y = end_y |
|
|
| |
| relay_disp = relay if relay else "(empty -> defaults)" |
| if relay and director_variant == "LTXDirector": |
| relay_disp = f"{relay}\n\nINERT: active Director is upstream LTXDirector." |
| y = _draw_text_box( |
| draw, x, y, inner_w, "Knob state", |
| relay_disp, ACCENT_RUN, max_lines=6, char_per_line=42, |
| keep_blank_lines=True, |
| ) |
| note_text = info_body if info_body.strip() else "(no note this render)" |
| y = _draw_text_box( |
| draw, x, y, inner_w, "Base notes", |
| note_text, ACCENT_RUN, max_lines=18, char_per_line=42, |
| keep_blank_lines=True, |
| ) |
|
|
| |
| feedback_text = "\n".join(feedback_lines) if feedback_lines else "(no feedback)" |
| fb_lines = _wrap_text(feedback_text, 42)[:6] |
| body_h = (20 + max(24, 22 * len(fb_lines)) + 14 + 20 + 50) |
| cx, cy, cw, end_y = _draw_section( |
| draw, x, y, inner_w, ACCENT_OUT, "Post-render", body_h, |
| ) |
| draw.text((cx, cy), "FEEDBACK", font=F_TAG, fill=ACCENT_OUT) |
| cy += 20 |
| for line in fb_lines: |
| draw.text((cx, cy), line, font=F_NOTE, fill=DIM) |
| cy += 22 |
| cy += 6 |
| draw.text((cx, cy), "OUTCOME", font=F_TAG, fill=ACCENT_OUT) |
| cy += 20 |
|
|
| gap = 8 |
| chip_w = (cw - 2 * gap) // 3 |
| for idx, (label, key) in enumerate(( |
| ("Motion", "motion"), |
| ("Sync", "sync"), |
| ("Sharp", "sharp"), |
| )): |
| _draw_score_chip( |
| draw, cx + idx * (chip_w + gap), cy, |
| chip_w, label, scores.get(key), |
| ) |
| y = end_y |
|
|
| |
| final_h = y + PAD_BOTTOM + inset |
| img = img.crop((0, 0, W, final_h)) |
| out_draw = ImageDraw.Draw(img) |
| out_draw.rounded_rectangle( |
| [inset, inset, W - inset, final_h - inset], |
| radius=RADIUS, outline=BORDER, width=1, |
| ) |
|
|
| out_path.parent.mkdir(parents=True, exist_ok=True) |
| pnginfo = None |
| metadata = state.get("metadata") |
| if metadata: |
| pnginfo = PngInfo() |
| pnginfo.add_text( |
| "koolook_audio_loop", |
| json.dumps(metadata, ensure_ascii=False, sort_keys=True), |
| ) |
| img.save(str(out_path), pnginfo=pnginfo) |
| return out_path |
|
|
|
|
| |
| |
| |
|
|
|
|
| def _rebuild_state_from_run_dir(run_dir: Path) -> dict[str, Any]: |
| """Reconstruct the loop_audio state dict from an on-disk run |
| folder. Reads the run's frozen runNNN_workflow.json and walks it through |
| the same extraction helpers loop_audio.py uses live.""" |
| from loop_audio import ( |
| active_relay_overrides, |
| derive_audio_state, detect_upstream_whatdreamscost_version, |
| director_flavor, director_pin_tag, director_type, director_widget, |
| expected_output_tracking, extract_director, extract_multilines, |
| extract_setup_variables, find_dotenv, |
| first_multiline, load_config, load_dotenv, |
| parse_feedback, parse_timeline, pick_existing_path, |
| segment_prompt_mode, |
| DEFAULT_CONFIG_PATH, |
| ) |
|
|
| name_parts = run_dir.name.split("_", 1) |
| m = re.match(r"run-(\d+)$", name_parts[0]) |
| nnn = int(m.group(1)) if m else 0 |
| label = name_parts[1] if len(name_parts) > 1 else "" |
| tagged_workflow = run_dir / f"run{nnn:03d}_workflow.json" |
| wf_path = tagged_workflow if tagged_workflow.is_file() else run_dir / "workflow.json" |
| if not wf_path.is_file(): |
| print(f"missing run workflow JSON in {run_dir}", file=sys.stderr) |
| sys.exit(2) |
| with wf_path.open(encoding="utf-8-sig") as f: |
| wf = json.load(f) |
| nodes = wf.get("nodes") or [] |
|
|
| |
| |
| env_file = find_dotenv() |
| if env_file is not None: |
| load_dotenv(env_file) |
|
|
| cfg = load_config(DEFAULT_CONFIG_PATH) |
| multilines = extract_multilines(nodes, cfg["tracked_multilines"]) |
| setup_variables = extract_setup_variables( |
| nodes, cfg.get("tracked_setup_variables", {}) |
| ) |
| output_tracking = expected_output_tracking(nodes, multilines, setup_variables) |
| director_node = extract_director(nodes, wf.get("links") or []) |
| timeline = parse_timeline(director_node) |
| audio_src = derive_audio_state(director_node, timeline) |
| relay_overrides_raw = active_relay_overrides( |
| nodes, wf.get("links") or [], director_node |
| ) |
| active_multilines = dict(multilines) |
| active_multilines["relay_overrides"] = [relay_overrides_raw] |
| upstream_whatdreamscost_version = detect_upstream_whatdreamscost_version() |
| pin_tag = director_pin_tag(director_node, upstream_whatdreamscost_version) |
| existing_metadata = _read_existing_metadata(run_dir) |
| run_date = _preserved_run_date(run_dir, existing_metadata) |
| scores, feedback_lines = parse_feedback( |
| first_multiline(multilines, "overlay - feedback") |
| ) |
| patch_meta: dict[str, str] = {} |
| patch_path = run_dir / "patch_state.txt" |
| if patch_path.is_file(): |
| for line in patch_path.read_text(encoding="utf-8").splitlines(): |
| if ":" not in line: |
| continue |
| key, value = line.split(":", 1) |
| patch_meta[key.strip()] = value.strip() |
| last_dev_sync_audio, last_dev_sync_at = parse_dev_sync_audio_line( |
| patch_meta.get("Last dev-sync-audio", "") |
| ) |
| repo_build = { |
| "commit": last_dev_sync_audio, |
| "synced_at": last_dev_sync_at, |
| "scope": patch_meta.get("Sync scope tag", ""), |
| "worktree": patch_meta.get("Sync worktree", ""), |
| } |
| source_workflow_name = "" |
| notes_path = run_dir / "notes.md" |
| if notes_path.is_file(): |
| for line in notes_path.read_text(encoding="utf-8").splitlines(): |
| match = re.match(r"- Setup name:\s+`([^`]+)`", line) |
| if match: |
| source_workflow_name = f"{match.group(1)}.json" |
| break |
|
|
| workflow_name = wf_path.name |
| state = { |
| "run_number": nnn, |
| "run_label": label, |
| "date": run_date, |
| "workflow_name": workflow_name, |
| "source_workflow_name": source_workflow_name, |
| "name": first_multiline(multilines, "name").strip() or "(unnamed)", |
| "relay_overrides_raw": relay_overrides_raw, |
| "info_body": first_multiline(multilines, "overlay - info").rstrip(), |
| "feedback_lines": feedback_lines, |
| "scores": scores, |
| "work_folder": pick_existing_path( |
| multilines.get("working_folder") or [] |
| ), |
| "output_folder": output_tracking.get("folder", ""), |
| "output_name": output_tracking.get("name", ""), |
| "output_tracking": output_tracking, |
| "metadata": { |
| "schema": "koolook.audio_loop.card_metadata.v1", |
| "source": "rebuilt from frozen run folder", |
| "run": { |
| "capture_number": f"{nnn:03d}", |
| "label": label, |
| "date": run_date, |
| "workflow": workflow_name, |
| "source_workflow": source_workflow_name, |
| "archived_workflow": workflow_name, |
| "setup_name": ( |
| Path(source_workflow_name).stem |
| if source_workflow_name else wf_path.stem |
| ), |
| }, |
| "setup": { |
| "base_name": first_multiline(multilines, "name").strip(), |
| "working_folder": scrub_path_for_metadata(output_tracking.get("folder", "")), |
| "input_path_exr": scrub_path_for_metadata( |
| first_multiline(setup_variables, "input_path_exr") |
| ), |
| "global_version": first_multiline(setup_variables, "version"), |
| "global_run_offset": first_multiline(setup_variables, "run_offset"), |
| "relay_overrides": relay_overrides_raw.strip(), |
| }, |
| "output": { |
| **output_tracking, |
| "folder": scrub_path_for_metadata(output_tracking.get("folder", "")), |
| }, |
| "director": { |
| "type": director_type(director_node), |
| "flavor": director_flavor(director_node), |
| "pin_tag": pin_tag, |
| "upstream_whatdreamscost_version": upstream_whatdreamscost_version, |
| "audio_src": audio_src, |
| "epsilon": director_widget(director_node, "epsilon"), |
| "duration_frames": director_widget(director_node, "duration_frames"), |
| "duration_seconds": director_widget(director_node, "duration_seconds"), |
| "frame_rate": metadata_numeric_widget( |
| director_widget(director_node, "frame_rate") |
| ), |
| "segment_prompt_mode": segment_prompt_mode( |
| timeline.get("segments") or [] |
| ), |
| "video_segments": len(timeline.get("segments") or []), |
| "audio_segments": len(timeline.get("audioSegments") or []), |
| }, |
| "repo": { |
| **repo_metadata_from_build( |
| repo_build, |
| patch_meta.get("Fork dir status", ""), |
| ), |
| "main_sha": patch_meta.get("MAIN SHA", ""), |
| }, |
| }, |
| "director_node": director_node, |
| "director_variant": director_type(director_node), |
| "director_flavor": director_flavor(director_node), |
| "director_pin_tag": pin_tag, |
| "audio_src": audio_src, |
| "epsilon": director_widget(director_node, "epsilon"), |
| "frame_rate": metadata_numeric_widget( |
| director_widget(director_node, "frame_rate") |
| ), |
| "segments": timeline.get("segments") or [], |
| "audio_segments": timeline.get("audioSegments") or [], |
| "segment_prompt_mode": segment_prompt_mode( |
| timeline.get("segments") or [] |
| ), |
| } |
| state["notes_md"] = render_notes_md( |
| nnn, |
| Path(source_workflow_name) if source_workflow_name else wf_path, |
| active_multilines, |
| setup_variables, |
| director_node, |
| timeline, |
| audio_src, |
| state["info_body"], |
| feedback_lines, |
| scores, |
| output_tracking, |
| commit_sha=patch_meta.get("MAIN SHA", ""), |
| ) |
| return state |
|
|
|
|
| def main() -> int: |
| p = argparse.ArgumentParser(description=__doc__) |
| p.add_argument("run_dir", type=Path) |
| p.add_argument("--dry-run", action="store_true") |
| p.add_argument( |
| "--no-delivery", |
| action="store_true", |
| help="Skip copying the rendered card to the external cards folder.", |
| ) |
| args = p.parse_args() |
|
|
| run_dir = args.run_dir.expanduser().resolve() |
| if not run_dir.is_dir(): |
| print(f"run folder not found: {run_dir}", file=sys.stderr) |
| return 2 |
|
|
| |
| |
| state = _rebuild_state_from_run_dir(run_dir) |
|
|
| out = run_dir / "card.png" |
| if args.dry_run: |
| print(f"(dry-run) would write: {out}") |
| return 0 |
| render_audio_card(state, out) |
| metadata_path = run_dir / "metadata.json" |
| notes_path = run_dir / "notes.md" |
| metadata_path.write_text( |
| json.dumps( |
| state.get("metadata") or {}, |
| ensure_ascii=False, |
| indent=2, |
| sort_keys=True, |
| ) + "\n", |
| encoding="utf-8", |
| ) |
| notes_path.write_text(state.get("notes_md") or "", encoding="utf-8") |
| if args.no_delivery: |
| delivery_status = "(skipped)" |
| else: |
| delivery_status = copy_delivery_card( |
| out, |
| state.get("output_tracking") or {}, |
| state.get("run_number"), |
| overwrite=True, |
| ) |
| print(f"wrote: {out}") |
| print(f"wrote: {metadata_path}") |
| print(f"wrote: {notes_path}") |
| print(f"delivered: {delivery_status}") |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| sys.exit(main()) |
|
|
