832 lines
30 KiB
Python
832 lines
30 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
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
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Palette + geometry — kept in lockstep with scripts/make_card.py.
|
|
# --------------------------------------------------------------------------
|
|
|
|
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) # amber — per-render knob state
|
|
ACCENT_BASE = (109, 180, 255) # sky — locked / scene
|
|
ACCENT_OUT = (123, 207, 128) # green — post-render outcome
|
|
NOTE_BG = (12, 12, 12)
|
|
RADIUS = 14
|
|
SECTION_RADIUS = 8
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Fonts — same fallback chain + same sizes as scripts/make_card.py.
|
|
# --------------------------------------------------------------------------
|
|
|
|
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)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Drawing primitives — copied 1:1 from scripts/make_card.py.
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
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)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Renderer — pure state-in / PNG-out. The ``state`` dict is built by
|
|
# loop_audio._build_state_for_card. All values come from the two source
|
|
# families documented at the top of this module.
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
# Shared helpers live in loop_audio so the renderer and extractor agree
|
|
# byte-for-byte on path-wrapping rules and audio-overlap geometry.
|
|
# Importing here keeps the standalone CLI working (it adds scripts/ to
|
|
# sys.path before invoking render_audio_card).
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from loop_audio import ( # type: ignore[import-not-found] # noqa: E402
|
|
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 {}
|
|
# NOTE: duration_frames / duration_seconds intentionally not read here
|
|
# — the dropped "Duration" row used them; segment time ranges convey
|
|
# the same info now. They stay in the state dict for notes.md.
|
|
|
|
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
|
|
# Wider key column so multi-word labels (relay_overrides, Working
|
|
# folder) don't collide with their values.
|
|
key_w = 178
|
|
|
|
# ----- HEADER -----
|
|
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)
|
|
|
|
# ----- BASE / RUN -----
|
|
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
|
|
|
|
# ----- DIRECTOR READINGS -----
|
|
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
|
|
|
|
# ----- AMBER NOTES -----
|
|
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,
|
|
)
|
|
|
|
# ----- POST-RENDER (green) — FEEDBACK + OUTCOME -----
|
|
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
|
|
|
|
# ----- crop to actual content + save -----
|
|
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
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Standalone CLI — re-render a card from an existing run folder.
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
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 ( # type: ignore[import-not-found]
|
|
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 []
|
|
|
|
# Load env so any env-var lookups inside the helpers (none right
|
|
# now, but a defensive match for make_card.py's pattern) work.
|
|
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
|
|
|
|
# sys.path already pointed at scripts/ by the module-level import of
|
|
# wrap_path/video_segment_has_audio above; no need to repeat it here.
|
|
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())
|