"""One-command demo recorder + real-model end-to-end validation.
YUI_BRAIN_BACKEND=transformers HF_HUB_DISABLE_XET=1 \
uv run --group train --group viz python scripts/record_demo.py
Boots ONE curated home, runs all four scripted cases through the REAL two-stage
models on CPU, asserts each one's outcome (channel + home-state change), speaks
Yui's reply with the real TTS, and stitches a per-frame "live pipeline trace"
(the home grid + the step indicator + captions, screenshotted from the real app
HTML) into data/video/yui-demo.mp4 — plus a few PNG stills and a short GIF for
the README / social post. No manual screen-recording; nothing mocked.
Renderer: Playwright (faithful to the app's own _render_home_html / _render_stepper)
when chromium is available, else a PIL fallback that draws the same grid + stepper
+ captions. Either way ffmpeg stitches frames + the per-case TTS wav.
This script intentionally drives _render_stepper directly per frame: the live app
streams the stepper mid-handler, but here we own the frame sequence, so we step it
explicitly so the recorded trace visibly progresses
Heard "Hey Yui" -> Listening -> Transcribing -> Routing (stage 1)
-> [Writing automation (stage 2), automation case only]
-> Acting on the home -> Done.
"""
from __future__ import annotations
import os
import shutil
import subprocess
import sys
import wave
from dataclasses import dataclass, field
from pathlib import Path
# Ensure the repo root is importable when run as `python scripts/record_demo.py`.
_REPO = Path(__file__).resolve().parent.parent
if str(_REPO) not in sys.path:
sys.path.insert(0, str(_REPO))
import yaml # noqa: E402
# Reuse the app's real rendering + routing helpers — do NOT reimplement them.
import app as appmod # noqa: E402
from voice.home_view import HomeView # noqa: E402
# ---------------------------------------------------------------------------
# Tunable constants (timings/captions live here, per the spec).
# ---------------------------------------------------------------------------
DEMO_HOME = _REPO / "data" / "homes" / "demo-curated.yaml"
OUT_DIR = _REPO / "data" / "video"
FRAMES_DIR = OUT_DIR / "_frames"
MP4_PATH = OUT_DIR / "yui-demo.mp4"
GIF_PATH = OUT_DIR / "yui-demo.gif"
FPS = 30 # output video frame rate
FRAME_W, FRAME_H = 1280, 720 # 16:9, social-friendly
STEP_HOLD_S = 2.2 # how long each pipeline step stays on screen
RESULT_HOLD_S = 4.0 # extra hold on the final (Done) frame so it reads
INTRO_HOLD_S = 2.5 # opening title card
GIF_FPS = 10 # smaller, looping highlight for README/social
def _pick_ffmpeg(name: str) -> str:
"""Prefer a full-featured ffmpeg/ffprobe: a system build with libx264 + a
modern amix beats the miniconda 4.3 build (no x264, old amix). Honor an
explicit override first, then probe candidates, then PATH."""
override = os.environ.get(f"YUI_{name.upper()}")
if override:
return override
candidates = [f"/usr/bin/{name}", shutil.which(name), f"/home/roton/miniconda3/bin/{name}"]
if name == "ffmpeg":
for c in candidates:
if c and Path(c).exists():
try:
enc = subprocess.run([c, "-hide_banner", "-encoders"],
capture_output=True, text=True).stdout
if "libx264" in enc:
return c
except Exception:
continue
for c in candidates:
if c and Path(c).exists():
return c
return name
FFMPEG = _pick_ffmpeg("ffmpeg")
FFPROBE = _pick_ffmpeg("ffprobe")
# A fixed "current time" so the scripted run is deterministic (also keeps the
# automation/time logic stable across re-records).
DEMO_TIME = "7:15 PM on Monday, June 15 2026"
@dataclass
class Case:
key: str
title: str
utterance: str
automation: bool = False # does this case run stage-2?
# filled in at runtime:
spoken: str = ""
passed: bool = False
detail: str = ""
raw_model: str = "" # raw stage-1/stage-2 model output (for reporting)
channel: str = ""
frames: list = field(default_factory=list)
CASES = [
Case("reject", "Case 1 - says no (unknown device)", "open the garage"),
Case("response", "Case 2 - answers a question", "is the front door locked?"),
Case("intent", "Case 3 - turns lights off", "turn off the kitchen light"),
Case(
"automation",
"Case 4 - writes an automation",
"turn on the porch light when motion is detected",
automation=True,
),
]
# ---------------------------------------------------------------------------
# Real pipeline per case (reuses app routing/firing; returns channel + reply +
# a human result line + the assertion verdict). No mocks.
# ---------------------------------------------------------------------------
def _snapshot_states(home: HomeView) -> dict[str, str]:
return {eid: st for eid, (st, _a) in home.home.snapshot().items()}
def run_case(home: HomeView, case: Case, renderer=None) -> None:
"""Run one case through the real models against `home`, mutate `case` with the
channel / spoken reply / PASS-FAIL verdict / raw model output, and a result
line for the caption. Reuses app._summarize_changes / _route_automation paths
by calling the same helpers the app does (stage-1 brain, then channel logic).
If `renderer` is given, a frame is captured at each pipeline step AS the case
runs — so the pre-action steps (Heard … Routing/Writing) show the home before
execution and the Acting/Done steps show it after. This makes the recorded
trace faithful: the grid actually changes when 'Acting on the home' lights up.
"""
from voice.brain import get_brain
def cap(step: str):
if renderer is not None:
_capture_frame(renderer, case, step, home)
devices = home.devices_block()
before = _snapshot_states(home)
# Pre-inference steps share the pre-action home state.
cap(appmod.STEP_HEARD)
cap(appmod.STEP_LISTENING)
cap(appmod.STEP_TRANSCRIBING)
cap(appmod.STEP_ROUTING)
# --- stage 1 (real model, local CPU) ---
result = get_brain().infer(devices, case.utterance, DEMO_TIME)
case.raw_model = repr(result)
if not isinstance(result, dict):
case.channel = "?"
case.detail = f"non-dict result: {result!r}"
case.spoken = "Sorry, something went wrong."
case.passed = False
cap(appmod.STEP_ACTING)
cap(appmod.STEP_DONE)
return
channel = next((k for k in ("intents", "response", "automation") if k in result), "?")
case.channel = channel
if channel == "intents":
cap(appmod.STEP_ACTING)
results = home.execute_intents(result)
summary = appmod._summarize_changes(home, results)
case.detail = summary
case.spoken = "Done." if "No device state changed" in summary else "Okay, done."
elif channel == "response":
cap(appmod.STEP_ACTING)
text = str(result["response"])
case.detail = f"Yui: {text}"
case.spoken = text
elif channel == "automation":
_run_automation(home, case, result, devices, cap)
else:
cap(appmod.STEP_ACTING)
case.detail = f"no known channel in {result!r}"
case.spoken = "Sorry, I didn't understand that."
after = _snapshot_states(home)
case.passed, verdict = _assert_case(case, before, after, channel)
case.detail = (case.detail + "\n\n" + verdict).strip()
cap(appmod.STEP_DONE)
def _run_automation(home: HomeView, case: Case, result: dict, devices: str, cap=None) -> None:
"""Stage-2: write YAML, load it, fire the trigger, animate — mirrors
app._route_automation but without the Gradio plumbing."""
from voice.brain import get_brain2
def _cap(step):
if cap is not None:
cap(step)
auto = result.get("automation")
request = auto.strip() if isinstance(auto, str) and auto.strip() else case.utterance
_cap(appmod.STEP_WRITING) # stage-2 about to run; home still in pre-fire state
body = get_brain2().write(devices, request)
case.raw_model = case.raw_model + "\n\n[stage-2 YAML]\n" + (body or "")
try:
conf = yaml.safe_load(body)
except yaml.YAMLError as exc:
conf = None
case._yaml_err = str(exc) # type: ignore[attr-defined]
if isinstance(conf, list) and len(conf) == 1:
conf = conf[0]
case._conf = conf # type: ignore[attr-defined]
case._yaml = body # type: ignore[attr-defined]
has_action = isinstance(conf, dict) and ("action" in conf or "actions" in conf)
has_trigger = isinstance(conf, dict) and ("trigger" in conf or "triggers" in conf)
if not (has_action and has_trigger):
_cap(appmod.STEP_ACTING)
case.detail = "Drafted YAML but it isn't a valid automation."
case.spoken = "I couldn't build a valid automation."
return
err = home.home.load_automation(conf)
if err:
_cap(appmod.STEP_ACTING)
case.detail = f"Couldn't register: {err}"
case.spoken = "That automation didn't load."
return
_cap(appmod.STEP_ACTING) # registered; about to fire the trigger + animate
fired = appmod._fire_automation(home, conf)
case.detail = "Registered the automation.\n" + fired
case.spoken = "Registered the automation."
def _assert_case(
case: Case, before: dict[str, str], after: dict[str, str], channel: str
) -> tuple[bool, str]:
"""Validate the case's outcome (channel + home-state change). Returns
(passed, human verdict line)."""
changed = {k: (before.get(k), after[k]) for k in after if before.get(k) != after[k]}
if case.key == "reject":
# Should decline via the response channel and NOT actuate a real device.
ok = channel == "response" and not changed
why = (
f"channel={channel} (want response), device changes={changed or 'none'}"
)
return ok, ("PASS - declined, no device touched" if ok else f"FAIL - {why}")
if case.key == "response":
ok = channel == "response"
return ok, (
"PASS - answered on the response channel"
if ok else f"FAIL - landed on {channel}, wanted response"
)
if case.key == "intent":
ok = channel == "intents" and before.get("light.kitchen") == "on" \
and after.get("light.kitchen") == "off"
return ok, (
"PASS - light.kitchen on -> off"
if ok else
f"FAIL - channel={channel}, light.kitchen "
f"{before.get('light.kitchen')} -> {after.get('light.kitchen')}"
)
if case.key == "automation":
conf = getattr(case, "_conf", None)
loaded = isinstance(conf, dict)
ok = (
channel == "automation"
and loaded
and before.get("light.porch") == "off"
and after.get("light.porch") == "on"
)
return ok, (
"PASS - YAML parsed + loaded; light.porch off -> on on motion"
if ok else
f"FAIL - channel={channel}, yaml_parsed={loaded}, light.porch "
f"{before.get('light.porch')} -> {after.get('light.porch')}"
)
return False, "FAIL - unknown case"
# ---------------------------------------------------------------------------
# Frame composition: each frame is the real app HTML (home grid + stepper) wrapped
# in a caption card. We snapshot the home AFTER the case ran, so the "Acting/Done"
# frames show the post-action state (lights flipped) — exactly what the live app
# shows once the .then handler returns.
# ---------------------------------------------------------------------------
_PAGE_CSS = """
* { box-sizing: border-box; }
body {
margin: 0; width: __W__px; height: __H__px;
font-family: 'Quicksand','Segoe UI',system-ui,sans-serif;
color: #2a2540;
background:
radial-gradient(900px 540px at 12% -8%, rgba(196,181,253,.55), transparent 60%),
radial-gradient(820px 500px at 96% 0%, rgba(244,182,255,.50), transparent 60%),
linear-gradient(180deg,#faf5ff 0%,#f5f3ff 55%,#fff 100%);
}
.wrap { padding: 28px 40px; height: 100%; display: flex; flex-direction: column; gap: 16px; }
.title { text-align: center; }
.title h1 {
margin: 0; font-size: 30px;
background: linear-gradient(90deg,#a855f7,#ec4899,#818cf8);
-webkit-background-clip: text; background-clip: text; color: transparent;
}
.title .case { font-size: 15px; opacity: .65; margin-top: 2px; }
.stepper-box {
background: rgba(255,255,255,.7); border: 1px solid #ece6fb;
border-radius: 14px; padding: 12px 10px;
}
.home-box {
flex: 1; background: rgba(255,255,255,.78); border: 1px solid #ece6fb;
border-radius: 18px; padding: 14px 20px; overflow: hidden;
}
.home-box h3 { margin: 4px 0 10px; font-size: 20px; }
.caption {
background: rgba(255,255,255,.85); border: 1px solid #ece6fb;
border-radius: 14px; padding: 12px 18px; font-size: 16px; line-height: 1.45;
min-height: 64px;
}
.caption .you { color: #6d28d9; font-weight: 700; }
.caption .yui { color: #be185d; font-weight: 700; }
/* the app's grid uses CSS vars; provide light values so it renders standalone */
:root {
--block-background-fill: #fff;
--border-color-primary: #e7e2f3;
--body-text-color: #2a2540;
}
"""
def _page_css() -> str:
return _PAGE_CSS.replace("__W__", str(FRAME_W)).replace("__H__", str(FRAME_H))
def _frame_html(case: Case, step: str, home: HomeView, caption_html: str) -> str:
# The app HTML carries a 🏠 emoji; this offline chromium has no emoji font, so
# strip it to avoid tofu boxes (purely cosmetic; we don't touch app.py).
home_html = appmod._render_home_html(home).replace("🏠 ", "")
stepper_html = appmod._render_stepper(step)
css = _page_css()
return (
""
"
"
"
Yui
"
f"
{case.title}
"
f"
{stepper_html}
"
f"
{home_html}
"
f"
{caption_html}
"
"
"
)
def _caption(case: Case, step: str) -> str:
you = f"You: “{_esc(case.utterance)}”"
if step == appmod.STEP_DONE and case.spoken:
# Show only the first result line of the detail (keeps the caption tidy),
# stripped of the markdown the app helpers emit (**bold**, leading "- ").
first = case.detail.splitlines()[0] if case.detail else ""
first = first.lstrip("- ").replace("**", "")
yui = f" Yui: “{_esc(case.spoken)}”"
extra = f" {_esc(first)}" if first else ""
return you + yui + extra
labels = {
appmod.STEP_HEARD: "Heard the wake word…",
appmod.STEP_LISTENING: "Listening…",
appmod.STEP_TRANSCRIBING: "Transcribing…",
appmod.STEP_ROUTING: "Routing with stage 1…",
appmod.STEP_WRITING: "Writing the automation with stage 2…",
appmod.STEP_ACTING: "Acting on the home…",
}
sub = labels.get(step, "")
return you + (f" {sub}" if sub else "")
def _esc(s: str) -> str:
import html
return html.escape(s or "")
# ---------------------------------------------------------------------------
# Renderers: Playwright (preferred) or PIL (fallback). Both expose render(html, path).
# ---------------------------------------------------------------------------
class PlaywrightRenderer:
def __init__(self):
from playwright.sync_api import sync_playwright
self._pw = sync_playwright().start()
self._browser = self._pw.chromium.launch()
self._page = self._browser.new_page(
viewport={"width": FRAME_W, "height": FRAME_H},
device_scale_factor=1,
)
def render(self, html: str, path: Path) -> None:
self._page.set_content(html, wait_until="networkidle")
self._page.screenshot(path=str(path), clip={"x": 0, "y": 0, "width": FRAME_W, "height": FRAME_H})
def close(self):
try:
self._browser.close()
self._pw.stop()
except Exception:
pass
class PILRenderer:
"""Fallback: draw the home as a labeled icon grid + the stepper + caption.
Doesn't parse the app HTML; redraws the same information so the frame is clean
even where chromium can't run."""
def __init__(self):
from PIL import ImageFont
self._ImageFont = ImageFont
self._font_lg = self._font(30)
self._font_md = self._font(18)
self._font_sm = self._font(15)
def _font(self, size):
for p in (
"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
"/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf",
):
if Path(p).exists():
return self._ImageFont.truetype(p, size)
return self._ImageFont.load_default()
# PIL renderer keeps the case/home/step in render_frame; this signature is
# adapted in build_frames below (it calls a PIL-specific path).
def render(self, html: str, path: Path) -> None: # pragma: no cover - unused path
raise RuntimeError("PILRenderer renders from structured data, not HTML")
def render_data(self, case: Case, step: str, home: HomeView, path: Path) -> None:
from PIL import Image, ImageDraw
img = Image.new("RGB", (FRAME_W, FRAME_H), (250, 245, 255))
d = ImageDraw.Draw(img)
# title
d.text((FRAME_W // 2, 30), "Yui", font=self._font_lg, fill=(168, 85, 247), anchor="mm")
d.text((FRAME_W // 2, 62), case.title, font=self._font_sm, fill=(120, 110, 140), anchor="mm")
# stepper
steps = appmod.STEPS
idx = steps.index(step) if step in steps else -1
x = 40
y = 100
for i, label in enumerate(steps):
mark = "✓ " if i < idx else ""
fill = (236, 72, 153) if i == idx else (
(90, 80, 110) if i < idx else (180, 175, 195)
)
txt = mark + label
d.text((x, y), txt, font=self._font_sm, fill=fill)
w = d.textlength(txt, font=self._font_sm)
x += w + 18
if x > FRAME_W - 160 and i < len(steps) - 1:
x = 40
y += 26
# home grid
gy = y + 44
d.text((40, gy - 6), f"🏠 {home.name}", font=self._font_md, fill=(42, 37, 64))
views = home.render()
cols = 5
cw, ch = (FRAME_W - 80) // cols, 96
for n, v in enumerate(views):
cx = 40 + (n % cols) * cw
cy = gy + 28 + (n // cols) * (ch + 12)
col = tuple(int(v.color.lstrip("#")[i:i + 2], 16) for i in (0, 2, 4))
d.rounded_rectangle([cx, cy, cx + cw - 12, cy + ch], radius=14,
outline=(231, 226, 243), width=2, fill=(255, 255, 255))
d.ellipse([cx + cw / 2 - 16, cy + 12, cx + cw / 2 + 16, cy + 44],
fill=col if v.active else (200, 200, 210))
d.text((cx + cw / 2 - 6, cy + 58), v.name[:16], font=self._font_sm,
fill=(42, 37, 64), anchor="ma")
d.text((cx + cw / 2 - 6, cy + 76), v.detail[:16], font=self._font_sm,
fill=(140, 135, 160), anchor="ma")
# caption
cap_y = FRAME_H - 96
d.rounded_rectangle([32, cap_y, FRAME_W - 32, FRAME_H - 24], radius=14,
outline=(231, 226, 243), width=2, fill=(255, 255, 255))
d.text((52, cap_y + 14), f"You: “{case.utterance}”",
font=self._font_md, fill=(109, 40, 217))
if step == appmod.STEP_DONE and case.spoken:
d.text((52, cap_y + 44), f"Yui: “{case.spoken}”",
font=self._font_md, fill=(190, 24, 93))
img.save(path)
def close(self):
pass
# ---------------------------------------------------------------------------
# Frame capture: rendered INLINE as each case runs (so pre-action steps show the
# pre-action home and Acting/Done show the post-action home — a faithful trace).
# ---------------------------------------------------------------------------
_frame_idx = 0 # monotonic across the whole run, so filenames stay ordered
def _next_frame_path() -> Path:
global _frame_idx
FRAMES_DIR.mkdir(parents=True, exist_ok=True)
p = FRAMES_DIR / f"f{_frame_idx:05d}.png"
_frame_idx += 1
return p
def _capture_frame(renderer, case: Case, step: str, home: HomeView) -> None:
"""Render one frame for `case` at pipeline `step` from the home's CURRENT live
state, append (step, path) to the case's frame list."""
path = _next_frame_path()
if isinstance(renderer, PILRenderer):
renderer.render_data(case, step, home, path)
else:
html = _frame_html(case, step, home, _caption(case, step))
renderer.render(html, path)
case.frames.append((step, path))
def render_intro(renderer) -> Path:
"""Render the opening title card and return its path."""
path = _next_frame_path()
if isinstance(renderer, PILRenderer):
from PIL import Image, ImageDraw
img = Image.new("RGB", (FRAME_W, FRAME_H), (250, 245, 255))
d = ImageDraw.Draw(img)
d.text((FRAME_W // 2, FRAME_H // 2 - 40), "Yui",
font=renderer._font(64), fill=(168, 85, 247), anchor="mm")
d.text((FRAME_W // 2, FRAME_H // 2 + 30),
"a local two-stage home assistant",
font=renderer._font_md, fill=(120, 110, 140), anchor="mm")
img.save(path)
else:
renderer.render(_intro_html(), path)
return path
def _intro_html() -> str:
css = _page_css()
return (
""
"
"
"
"
"
Yui
"
"
"
"a local two-stage home assistant — it routes, answers, acts, and "
"writes Home Assistant automations
"
"
"
)
# ---------------------------------------------------------------------------
# TTS per case + audio assembly.
# ---------------------------------------------------------------------------
def _wav_duration(path: str) -> float:
try:
with wave.open(path, "rb") as wf:
return wf.getnframes() / float(wf.getframerate())
except Exception:
return 0.0
def synth_replies(cases: list[Case]) -> dict[str, str]:
"""Synthesize each case's spoken reply. Returns {case.key: wav path or ''}."""
from voice import tts
out = {}
for case in cases:
path = tts.synth(case.spoken) if case.spoken else None
out[case.key] = path or ""
return out
# ---------------------------------------------------------------------------
# ffmpeg stitching. Build a concat of per-frame stills with per-case timing, then
# overlay a per-case audio track aligned to the Done frame.
# ---------------------------------------------------------------------------
def _ffmpeg_codec_args() -> list[str]:
"""Prefer libx264 (portable H.264); fall back to mpeg4 if x264 is missing."""
try:
enc = subprocess.run(
[FFMPEG, "-hide_banner", "-encoders"], capture_output=True, text=True
).stdout
except Exception:
enc = ""
if "libx264" in enc:
return ["-c:v", "libx264", "-pix_fmt", "yuv420p"]
return ["-c:v", "mpeg4", "-q:v", "3", "-pix_fmt", "yuv420p"]
def build_video(cases: list[Case], audio: dict[str, str], intro: Path | None) -> dict:
"""Assemble frames + per-case audio into MP4_PATH. Returns timing info.
Strategy: write a frame-list (ffmpeg concat demuxer) holding each PNG for its
on-screen duration, and a parallel silent/spoken audio timeline so each case's
TTS starts when its Done frame appears. We render the audio timeline as a single
wav (silence + each reply at the right offset) via ffmpeg's adelay+amix, then mux.
"""
OUT_DIR.mkdir(parents=True, exist_ok=True)
# 1) Build the visual concat list with per-frame durations, tracking the
# timeline offset where each case's Done frame begins (for audio sync).
concat_lines = []
t = 0.0
audio_offsets = {} # case.key -> seconds at which its Done frame appears
if intro:
concat_lines.append(f"file '{intro}'")
concat_lines.append(f"duration {INTRO_HOLD_S}")
t += INTRO_HOLD_S
for case in cases:
for step, path in case.frames:
hold = RESULT_HOLD_S if step == appmod.STEP_DONE else STEP_HOLD_S
if step == appmod.STEP_DONE:
audio_offsets[case.key] = t
concat_lines.append(f"file '{path}'")
concat_lines.append(f"duration {hold}")
t += hold
# concat demuxer needs the last file repeated (its final duration is ignored).
last = cases[-1].frames[-1][1]
concat_lines.append(f"file '{last}'")
total = t
concat_txt = FRAMES_DIR / "concat.txt"
concat_txt.write_text("\n".join(concat_lines) + "\n")
codec = _ffmpeg_codec_args()
# 2) Build the audio timeline: each reply delayed to its Done offset, amix'd.
have_audio = [(c.key, audio.get(c.key, ""), audio_offsets.get(c.key, 0.0))
for c in cases if audio.get(c.key)]
audio_path = None
if have_audio:
audio_path = FRAMES_DIR / "track.wav"
inputs = []
filters = []
for i, (_k, wav, off) in enumerate(have_audio):
inputs += ["-i", wav]
ms = int(off * 1000)
filters.append(f"[{i}:a]adelay={ms}|{ms}[a{i}]")
mix = "".join(f"[a{i}]" for i in range(len(have_audio)))
fc = ";".join(filters) + f";{mix}amix=inputs={len(have_audio)}:normalize=0[out]"
cmd = [
FFMPEG, "-hide_banner", "-loglevel", "error", "-y", *inputs,
"-filter_complex", fc, "-map", "[out]",
"-t", f"{total:.3f}", str(audio_path),
]
subprocess.run(cmd, check=True)
# 3) Mux frames (+audio) into the final mp4.
cmd = [
FFMPEG, "-hide_banner", "-loglevel", "error", "-y",
"-f", "concat", "-safe", "0", "-i", str(concat_txt),
]
if audio_path:
cmd += ["-i", str(audio_path)]
cmd += ["-r", str(FPS), *codec]
if audio_path:
cmd += ["-c:a", "aac", "-shortest"]
cmd += [str(MP4_PATH)]
subprocess.run(cmd, check=True)
return {"duration": total, "audio": bool(audio_path)}
def build_gif(cases: list[Case]) -> None:
"""A short looping GIF: the Done frame of each case (the payoff stills)."""
done_frames = []
for c in cases:
done = next((p for s, p in c.frames if s == appmod.STEP_DONE), None)
if done:
done_frames.append(done)
if not done_frames:
return
listfile = FRAMES_DIR / "gif_concat.txt"
lines = []
for p in done_frames:
lines.append(f"file '{p}'")
lines.append("duration 1.6")
lines.append(f"file '{done_frames[-1]}'")
listfile.write_text("\n".join(lines) + "\n")
# palette for a clean GIF at a smaller width.
gw = 720
vf = f"fps={GIF_FPS},scale={gw}:-1:flags=lanczos,split[s0][s1];[s0]palettegen[p];[s1][p]paletteuse"
cmd = [
FFMPEG, "-hide_banner", "-loglevel", "error", "-y",
"-f", "concat", "-safe", "0", "-i", str(listfile),
"-vf", vf, "-loop", "0", str(GIF_PATH),
]
subprocess.run(cmd, check=True)
def export_stills(cases: list[Case]) -> list[Path]:
"""Copy 3 representative Done frames out as named PNG stills."""
stills = []
pick = {"intent": "still-intent.png", "automation": "still-automation.png",
"response": "still-response.png"}
for c in cases:
if c.key in pick:
done = next((p for s, p in c.frames if s == appmod.STEP_DONE), None)
if done:
dst = OUT_DIR / pick[c.key]
shutil.copyfile(done, dst)
stills.append(dst)
return stills
# ---------------------------------------------------------------------------
_HOME: HomeView | None = None # the single booted home (spec: boot ONCE)
def main() -> int:
global _HOME
os.environ.setdefault("YUI_BRAIN_BACKEND", "transformers")
os.environ.setdefault("HF_HUB_DISABLE_XET", "1")
print("=" * 64)
print("Yui demo recorder — real two-stage pipeline (CPU) + video render")
print("=" * 64)
# Boot ONE home (the harness leaks threads per boot — boot once, run cases
# sequentially against it).
print(f"[boot] loading curated home {DEMO_HOME.name} …", flush=True)
_HOME = HomeView()
_HOME.load(DEMO_HOME)
print(f"[boot] {len(_HOME.render())} devices: "
f"{', '.join(v.entity_id for v in _HOME.render())}", flush=True)
# ---- renderer selection (before the cases, so frames are captured inline as
# each case runs — pre-action steps show the pre-action home, Acting/Done show
# the post-action home: a faithful live trace) ----
if FRAMES_DIR.exists():
shutil.rmtree(FRAMES_DIR)
try:
renderer = PlaywrightRenderer()
renderer_name = "Playwright (chromium)"
except Exception as exc:
print(f"[render] Playwright unavailable ({exc}); falling back to PIL.", flush=True)
renderer = PILRenderer()
renderer_name = "PIL"
print(f"[render] renderer = {renderer_name}", flush=True)
intro_path = render_intro(renderer)
# Run each case through the REAL models, sequentially, asserting outcomes,
# capturing a frame at each pipeline step as it runs.
for case in CASES:
print(f"\n[case] {case.key}: “{case.utterance}”", flush=True)
try:
run_case(_HOME, case, renderer)
except Exception as exc: # never abort the whole run; report loudly
import traceback
case.passed = False
case.detail = f"EXCEPTION: {exc}"
case.raw_model = traceback.format_exc()
verdict = "PASS" if case.passed else "FAIL"
print(f" channel={case.channel} -> {verdict}", flush=True)
for line in case.detail.splitlines():
print(f" {line}", flush=True)
renderer.close()
# ---- validation summary (loud) ----
print("\n" + "=" * 64)
print("VALIDATION SUMMARY")
print("=" * 64)
all_pass = True
for case in CASES:
status = "PASS ✅" if case.passed else "FAIL ❌"
all_pass = all_pass and case.passed
print(f" [{status}] {case.key:11s} channel={case.channel:11s} — {case.title}")
if not case.passed:
print(f" raw model output: {case.raw_model[:400]}")
print("=" * 64)
# ---- TTS ----
print("\n[tts] synthesizing replies …", flush=True)
audio = synth_replies(CASES)
for c in CASES:
a = audio.get(c.key)
print(f" {c.key}: {'wav ' + a + f' ({_wav_duration(a):.1f}s)' if a else 'no audio'}",
flush=True)
nframes = sum(len(c.frames) for c in CASES) + 1
print(f"\n[render] {nframes} frames -> {FRAMES_DIR}", flush=True)
# ---- video + gif + stills ----
print("[ffmpeg] stitching mp4 …", flush=True)
info = build_video(CASES, audio, intro_path)
print("[ffmpeg] building gif …", flush=True)
build_gif(CASES)
stills = export_stills(CASES)
# ---- report artifacts ----
print("\n" + "=" * 64)
print("ARTIFACTS")
print("=" * 64)
if MP4_PATH.exists():
size = MP4_PATH.stat().st_size
dur = _probe_duration(MP4_PATH)
print(f" mp4: {MP4_PATH} ({size/1e6:.2f} MB, {dur:.1f}s, audio={info['audio']})")
else:
print(" mp4: MISSING ❌")
if GIF_PATH.exists():
print(f" gif: {GIF_PATH} ({GIF_PATH.stat().st_size/1e6:.2f} MB)")
for s in stills:
print(f" still: {s} ({s.stat().st_size/1e3:.0f} KB)")
print("=" * 64)
return 0 if all_pass else 1
def _probe_duration(path: Path) -> float:
try:
out = subprocess.run(
[FFPROBE, "-v", "error", "-show_entries", "format=duration",
"-of", "default=noprint_wrappers=1:nokey=1", str(path)],
capture_output=True, text=True,
).stdout.strip()
return float(out)
except Exception:
return 0.0
if __name__ == "__main__":
sys.exit(main())