| |
| """teaser · 编辑 case（外观编辑 + 动作 / 轨迹控制）的合成（maliang venv，CPU）。被 edit_app.py compose / edit_sameworld.py compose |
| 调用，也可单独跑。画法沿用 compose.py（深色底、细边框、胶囊标签、ffmpeg x264 crf 16 BT.709、16 fps、1920×1080）。 |
| |
| 产物（outputs/teaser/edit/app/showcase/，下称 $O） |
| look_<b>.mp4 同一张语义地图 × 多种首帧外观：左上 = 语义图，其余 = 原首帧 + 各外观编辑（同 ids、同按键） |
| look_all_<v>.mp4 同一种编辑（如 night）在所有底子上（每格右上角是原版小窗） |
| look_pairs.mp4 每个底子挑 2 种编辑，「原版 | 编辑」并排 4×… 宫格 |
| mirror_<b>.mp4 「原版 | 镜像（A↔D）」并排 + 按键 HUD |
| text_<b>.mp4 原版 + 只改文本的 3 种（全 standing still / 反方向 / 一直攻击）2×2 + 文本按键 HUD（语义不变） |
| sameworld_<g>.mp4 左 = 俯视总图（各 clip 轨迹一色 + 当前位置圆点），右 = 各 clip 的生成视频 + 按键 HUD |
| sheets/*.jpg 对应的帧拼图（帧 0/20/40/60/80），给人看 / 报告用 |
| 按键 HUD（keys_hud）：W / A / S / D 方向键 + ATK，按下 = 亮。帧 f 的按键 = blocks20[max(0, ceil(f/4) − 1)]（cell j = 帧 4j−3..4j |
| ← block j−1，cell 0 = cell 1）。 |
| |
| usage：$PY scripts/teaser/edit_compose.py app [--only look act] | sameworld <group>... |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import math |
| import os |
| import sys |
|
|
| import cv2 |
| import numpy as np |
| from PIL import Image, ImageDraw |
|
|
| M = "/opt/dlami/nvme/zhiyangdeng/maliang" |
| HERE = os.path.dirname(os.path.abspath(__file__)) |
| for _p in (f"{M}/code/gc25d", HERE): |
| if _p not in sys.path: |
| sys.path.insert(0, _p) |
| import compose as CP |
|
|
| A = f"{M}/outputs/teaser/edit/app" |
| O = f"{A}/showcase" |
| W, H = CP.CANVAS |
|
|
|
|
| |
| def block_of(f: int) -> int: |
| return max(0, math.ceil(f / 4) - 1) |
|
|
|
|
| _HUD: dict = {} |
|
|
|
|
| def keys_hud(state, size: int = 26): |
| """state = (w, a, s, d, attack) 0/1 → (rgb, alpha)：上排 W，下排 A S D，右边 ATK。按下 = 白底黑字，否则半透明黑底灰字。""" |
| key = (tuple(int(x) for x in state), size) |
| if key in _HUD: |
| return _HUD[key] |
| g = round(size * 0.18) |
| aw = round(size * 2.1) |
| wd, ht = 3 * size + 2 * g + g + aw, 2 * size + g |
| im = Image.new("RGBA", (wd + 2, ht + 2), (0, 0, 0, 0)) |
| d = ImageDraw.Draw(im) |
| pos = {"W": (size + g, 0), "A": (0, size + g), "S": (size + g, size + g), "D": (2 * (size + g), size + g)} |
| on = dict(zip("WASD", key[0][:4])) |
| fs = round(size * 0.55) |
| for k, (x, y) in pos.items(): |
| lit = on[k] |
| d.rounded_rectangle((x, y, x + size, y + size), radius=round(size * 0.22), |
| fill=(245, 245, 250, 240) if lit else (0, 0, 0, 120), |
| outline=(255, 255, 255, 200) if lit else (200, 200, 210, 110), width=1) |
| tw = CP.text_w(k, fs, "SemiBold") |
| CP.draw_text(d, x + (size - tw) / 2, y + (size - fs) / 2 - 1, k, fs, (20, 20, 25, 255) if lit else (210, 212, 220, 220), |
| "SemiBold") |
| x0 = 3 * size + 3 * g |
| lit = key[0][4] |
| d.rounded_rectangle((x0, size + g, x0 + aw, 2 * size + g), radius=round(size * 0.22), |
| fill=(255, 120, 70, 240) if lit else (0, 0, 0, 120), |
| outline=(255, 255, 255, 200) if lit else (200, 200, 210, 110), width=1) |
| tw = CP.text_w("ATK", fs, "SemiBold") |
| CP.draw_text(d, x0 + (aw - tw) / 2, size + g + (size - fs) / 2 - 1, "ATK", fs, |
| (20, 20, 25, 255) if lit else (210, 212, 220, 220), "SemiBold") |
| a = np.asarray(im, np.float32) |
| _HUD[key] = (a[..., :3].astype(np.uint8), a[..., 3:] / 255.0) |
| return _HUD[key] |
|
|
|
|
| def load_blocks(case_dir: str) -> np.ndarray | None: |
| p = f"{case_dir}/actions.npy" |
| if os.path.exists(p): |
| return np.load(p) |
| meta = json.load(open(f"{case_dir}/meta.json")) |
| act = (meta["job"].get("edit") or {}).get("actions") |
| return np.load(act) if act and os.path.exists(act) else None |
|
|
|
|
| |
| PLAYER_IDS, BOSS_IDS = (10, 12, 13, 14, 15, 16), (11, 17, 18, 19, 20, 21) |
|
|
|
|
| def zoom_boxes(ids: np.ndarray, target: str, factor: float, smooth: int = 7) -> list[tuple[int, int, int, int]]: |
| """逐帧跟拍框：target（player / boss）在 ids[t] 里的掩码质心（看不见就沿用上一帧），滑动平均 smooth 帧，框 = 画面 / factor。""" |
| F, h, w = ids.shape |
| cs, last = [], np.array([w / 2, h / 2]) |
| for t in range(F): |
| ys, xs = np.nonzero(np.isin(ids[t], PLAYER_IDS if target == "player" else BOSS_IDS)) |
| if len(xs) >= 30: |
| last = np.array([xs.mean(), ys.mean()]) |
| cs.append(last) |
| cs = np.asarray(cs) |
| k = np.ones(smooth) / smooth |
| cs = np.stack([np.convolve(np.pad(cs[:, i], smooth // 2, mode="edge"), k, mode="valid") for i in range(2)], 1) |
| bw, bh = w / factor, h / factor |
| out = [] |
| for cx, cy in cs: |
| x0 = int(round(np.clip(cx - bw / 2, 0, w - bw))) |
| y0 = int(round(np.clip(cy - bh / 2, 0, h - bh))) |
| out.append((x0, y0, x0 + int(bw), y0 + int(bh))) |
| return out |
|
|
|
|
| |
| def tiles_video(out: str, tiles: list[dict], cols: int, title: str | None = None, sub: str | None = None, |
| margin: int = 40, gap: int = 14, crf: int = 16, nf: int = 81) -> None: |
| """tiles：[{kind: gen|sem|img, case（dir）, label, hud（bool）, inset（dir：右上角小窗，取它的 gen）}]，同步播放。""" |
| top = 0 |
| if title: |
| top = 64 if sub else 48 |
| rows, tw, th, pos = CP.grid_layout(len(tiles), cols, W, H - top, 832 / 480, margin, gap) |
| pos = [(x, y + top) for x, y in pos] |
| data = [] |
| for t in tiles: |
| d = dict(t) |
| if t["kind"] == "gen": |
| d["frames"] = CP.read_video(f"{t['case']}/gen.mp4", nf) |
| d["blocks"] = load_blocks(t["case"]) if t.get("hud") else None |
| elif t["kind"] == "sem": |
| d["ids"] = CP.load_case(t["case"])["ids"]()[:nf] |
| if t.get("inset"): |
| d["inset_frames"] = CP.read_video(f"{t['inset']}/gen.mp4", nf) |
| if t.get("zoom") and t["kind"] == "gen": |
| ids = CP.load_case(t["case"])["ids"]()[:nf] |
| boxes = zoom_boxes(ids, *t["zoom"]) |
| d["frames"] = np.stack([cv2.resize(f[y0:y1, x0:x1], (f.shape[1], f.shape[0]), interpolation=cv2.INTER_LANCZOS4) |
| for f, (x0, y0, x1, y1) in zip(d["frames"], boxes)]) |
| data.append(d) |
| fs = min(24, max(14, round(th * 0.06))) |
| hud_size = min(28, max(16, round(th * 0.07))) |
| for d in data: |
| d["pill"] = CP.pill(d["label"], fs, alpha=0.62) if d.get("label") else None |
| base = Image.new("RGB", (W, H), CP.BG) |
| if title: |
| dr = ImageDraw.Draw(base) |
| CP.draw_text(dr, margin, 14, title, 28, CP.FG, "SemiBold") |
| if sub: |
| CP.draw_text(dr, margin, 14 + 34, sub, 18, CP.MUTED) |
| base = np.asarray(base).copy() |
| for x, y in pos: |
| CP.frame_rect(base, x, y, tw, th) |
|
|
| def frames(): |
| for i in range(nf): |
| fr = base.copy() |
| for d, (x, y) in zip(data, pos): |
| if d["kind"] == "gen": |
| tile = CP.fit(d["frames"][min(i, len(d["frames"]) - 1)], tw, th) |
| elif d["kind"] == "sem": |
| tile = CP.sem_panel(d["ids"][min(i, len(d["ids"]) - 1)], tw, th) |
| else: |
| tile = CP.fit(d["img"], tw, th) |
| fr[y:y + th, x:x + tw] = tile |
| if d.get("inset_frames") is not None: |
| iw, ih = tw // 4, th // 4 |
| fr[y + 6:y + 6 + ih, x + tw - iw - 6:x + tw - 6] = CP.fit(d["inset_frames"][i], iw, ih) |
| CP.frame_rect(fr, x + tw - iw - 6, y + 6, iw, ih, (235, 235, 240), 1) |
| if d["pill"] is not None: |
| CP.paste_rgba(fr, d["pill"], x + round(fs * 0.5), y + th - d["pill"][0].shape[0] - round(fs * 0.5)) |
| if d.get("blocks") is not None: |
| hud = keys_hud(d["blocks"][block_of(i)], hud_size) |
| CP.paste_rgba(fr, hud, x + tw - hud[0].shape[1] - 8, y + th - hud[0].shape[0] - 8) |
| yield fr |
| os.makedirs(os.path.dirname(out), exist_ok=True) |
| n = CP.write_video(out, frames(), crf=crf) |
| print(f"→ {out}（{len(tiles)} 格，{rows}×{cols}，{n} 帧）", flush=True) |
|
|
|
|
| def sheet(paths: list[str], labels: list[str], out: str, frames=(0, 20, 40, 60, 80), tw: int = 312, sem: str | None = None): |
| import edit_app as E |
| E.sheet(paths, labels, out, frames, tw) |
|
|
|
|
| |
| def compose_app(a) -> None: |
| import edit_app as E |
| only = set(a.only or ["look", "act", "all_look", "gallery"]) |
| R = f"{A}/roll" |
| bases = E.load_bases() |
| done = lambda n: os.path.exists(f"{R}/{n}/meta.json") |
| if "look" in only: |
| for s, b in bases.items(): |
| vs = [v for v in ["orig"] + b["variants"] if done(f"{s}__{v}")] |
| if len(vs) < 2: |
| continue |
| tiles = [{"kind": "sem", "case": f"{R}/{s}__orig", "label": "Semantic map (shared)"}] + \ |
| [{"kind": "gen", "case": f"{R}/{s}__{v}", "label": E.EDIT_EN[v]} for v in vs] |
| cols = 4 if len(tiles) > 9 else 3 |
| tiles_video(f"{O}/look_{s}.mp4", tiles, cols, "Same semantic map, different first frames", |
| "Layout from the semantic map · appearance from the first frame (edited with Qwen-Image) · " |
| "same keys") |
| sheet([f"{R}/{s}__{v}" for v in vs], [f"{s}: {E.EDIT_EN[v]}" for v in vs], f"{O}/sheets/look_{s}.jpg") |
| if "gallery" in only: |
| for s, who in E.GALLERY.items(): |
| vs = [v for v in ["orig"] + E.gallery_variants(s) if done(f"{s}__{v}")] |
| if len(vs) < 3: |
| continue |
| tiles = [{"kind": "sem", "case": f"{R}/{s}__orig", "label": "Semantic map (shared)"}] + \ |
| [{"kind": "gen", "case": f"{R}/{s}__{v}", "label": E.EDIT_EN[v]} for v in vs] |
| tiles_video(f"{O}/gallery_{s}_{who}.mp4", tiles, 4, f"Same map, different {'heroes' if who == 'hero' else 'bosses'}", |
| f"Only the {'hero' if who == 'hero' else 'boss'} in the first frame is edited (local Qwen-Image edit, " |
| "pixels outside the character mask pinned) · same semantic map and keys") |
| sheet([f"{R}/{s}__{v}" for v in vs], [f"{s}: {E.EDIT_EN[v]}" for v in vs], f"{O}/sheets/gallery_{s}_{who}.jpg") |
| zt = [{"kind": "gen", "case": f"{R}/{s}__{v}", "label": E.EDIT_EN[v], |
| "zoom": ("player" if who == "hero" else "boss", 2.6 if who == "hero" else 1.8)} for v in vs] |
| tiles_video(f"{O}/gallery_{s}_{who}_zoom.mp4", zt, 4, |
| f"Same map, different {'heroes' if who == 'hero' else 'bosses'} (tracking close-up)", |
| "Camera crop follows the character's semantic mask · identity comes from the edited first frame, " |
| "pose and motion from the semantic capsules") |
| if "all_look" in only: |
| allv = sorted({v for b in bases.values() for v in b["variants"]}) |
| for v in allv: |
| names = [s for s, b in bases.items() if v in b["variants"] and done(f"{s}__{v}")] |
| if len(names) < 3: |
| continue |
| tiles = [{"kind": "gen", "case": f"{R}/{s}__{v}", "label": f"{s}", "inset": f"{R}/{s}__orig"} |
| for s in names[:12]] |
| tiles_video(f"{O}/look_all_{v}.mp4", tiles, 4 if len(tiles) > 9 else 3, f"Edit: {E.EDIT_EN[v]}", |
| "Each tile: edited first frame + the original semantic map and keys (inset = original rollout)") |
| if "act" in only: |
| for s in E.MIRROR_BASES: |
| if done(f"{s}__orig") and done(f"{s}__mirror"): |
| tiles = [{"kind": "sem", "case": f"{R}/{s}__orig", "label": "Semantic map"}, |
| {"kind": "sem", "case": f"{R}/{s}__mirror", "label": "Mirrored semantic map"}, |
| {"kind": "gen", "case": f"{R}/{s}__orig", "label": "Original keys", "hud": True}, |
| {"kind": "gen", "case": f"{R}/{s}__mirror", "label": "Mirrored world, A/D swapped", |
| "hud": True}] |
| tiles_video(f"{O}/mirror_{s}.mp4", tiles, 2, "Mirror = action control", |
| "Semantic map + first frame mirrored left-right, keys A and D swapped: same world, opposite way") |
| sheet([f"{R}/{s}__orig", f"{R}/{s}__mirror"], [f"{s}: original", f"{s}: mirrored (A<->D)"], |
| f"{O}/sheets/mirror_{s}.jpg") |
| for s in E.TEXT_BASES: |
| names = [f"{s}__orig"] + [f"{s}__text_{t}" for t in E.TEXT_VARIANTS] |
| if all(done(n) for n in names): |
| labs = ["Original text"] + [E.TEXT_EN[t] for t in E.TEXT_VARIANTS] |
| tiles = [{"kind": "gen", "case": f"{R}/{n}", "label": l, "hud": True} for n, l in zip(names, labs)] |
| tiles_video(f"{O}/text_{s}.mp4", tiles, 2, "Only the action text changes", |
| "Same semantic map + same first frame; keys / prompts replaced (HUD = the text given)") |
| sheet([f"{R}/{n}" for n in names], [f"{s}: {l}" for l in labs], f"{O}/sheets/text_{s}.jpg") |
|
|
|
|
| |
| def compose_sameworld(name: str, a=None) -> None: |
| S = f"{A}/sameworld" |
| info = json.load(open(f"{S}/maps/{name}/group.json")) |
| cases = [f"{S}/roll/{name}__{c['k']}_{c['start']:06d}" for c in info["clips"]] |
| cases = [c for c in cases if os.path.exists(f"{c}/meta.json")] |
| if len(cases) < 2: |
| print(f"{name}：完成的 rollout 不够（{len(cases)}）") |
| return |
| clips = [c for c in info["clips"] if os.path.exists(f"{S}/roll/{name}__{c['k']}_{c['start']:06d}/meta.json")] |
| plan = cv2.cvtColor(cv2.imread(info["plan"]), cv2.COLOR_BGR2RGB) |
| nf = 81 |
| gens = [CP.read_video(f"{c}/gen.mp4", nf) for c in cases] |
| blocks = [load_blocks(c) for c in cases] |
| top = 64 |
| mg, gap = 36, 14 |
| |
| pts = np.concatenate([np.asarray(c["foot0_px"], np.float32) for c in clips] + |
| [np.asarray(c["path_px"], np.float32) for c in clips], 0) |
| lo, hi = pts.min(0), pts.max(0) |
| pad = 0.15 * (hi - lo).max() |
| x0c, y0c = int(max(0, lo[0] - pad)), int(max(0, lo[1] - pad)) |
| x1c, y1c = int(min(plan.shape[1], hi[0] + pad)), int(min(plan.shape[0], hi[1] + pad)) |
| plan = plan[y0c:y1c, x0c:x1c] |
| PLAN_W, PLAN_H = 560, H - top - 2 * mg |
| sc = min(PLAN_W / plan.shape[1], PLAN_H / plan.shape[0]) |
| pw, ph = round(plan.shape[1] * sc), round(plan.shape[0] * sc) |
| plan_s = cv2.resize(plan, (pw, ph), interpolation=cv2.INTER_AREA if sc < 1 else cv2.INTER_LINEAR) |
| py0 = top + (H - top - ph) // 2 |
| rx0 = mg + pw + gap |
| n = len(cases) |
| cols = 2 if n > 1 else 1 |
| rows, tw, th, pos = CP.grid_layout(n, cols, W - rx0, H - top, 832 / 480, mg, gap) |
| pos = [(x + rx0, y + top) for x, y in pos] |
| fs = min(24, max(14, round(th * 0.06))) |
| pills = [] |
| for c in clips: |
| col = tuple(int(v) for v in CP_COLORS[c["k"] % len(CP_COLORS)]) |
| pills.append(CP.pill(f"Play {c['k'] + 1}", fs, fg=(20, 20, 25), bg=col, alpha=0.9)) |
| mode = info.get("mode") |
| title = ("Same world, different playthroughs" if mode == "union" else "Same start frame, different keys") |
| sub = ("One procedural world built from the union of the clips' footprints · each clip: own camera, own keys, " |
| "own first frame (same scene prompt + seed)" if mode == "union" else |
| "Clips translated to one common start (boss removed) · one shared first frame · each clip: its own keys") |
| base = Image.new("RGB", (W, H), CP.BG) |
| dr = ImageDraw.Draw(base) |
| CP.draw_text(dr, mg, 14, title, 28, CP.FG, "SemiBold") |
| CP.draw_text(dr, mg, 48, sub, 17, CP.MUTED) |
| base = np.asarray(base).copy() |
| base[py0:py0 + ph, mg:mg + pw] = plan_s |
| CP.frame_rect(base, mg, py0, pw, ph) |
| for x, y in pos: |
| CP.frame_rect(base, x, y, tw, th) |
| hud_size = min(28, max(16, round(th * 0.07))) |
| paths = [(np.asarray(c["path_px"], np.float32) - [x0c, y0c]) * sc for c in clips] |
|
|
| def frames(): |
| for i in range(nf): |
| fr = base.copy() |
| for k, (c, P) in enumerate(zip(clips, paths)): |
| j = min(len(P) - 1, round(i * (len(P) - 1) / (nf - 1))) |
| col = tuple(int(v) for v in CP_COLORS[c["k"] % len(CP_COLORS)]) |
| cx, cy = int(round(mg + P[j][0])), int(round(py0 + P[j][1])) |
| cv2.circle(fr, (cx, cy), 9, (0, 0, 0), -1, cv2.LINE_AA) |
| cv2.circle(fr, (cx, cy), 7, col, -1, cv2.LINE_AA) |
| for k, (g, b, (x, y)) in enumerate(zip(gens, blocks, pos)): |
| fr[y:y + th, x:x + tw] = CP.fit(g[min(i, len(g) - 1)], tw, th) |
| CP.paste_rgba(fr, pills[k], x + round(fs * 0.5), y + round(fs * 0.5)) |
| if b is not None: |
| hud = keys_hud(b[block_of(i)], hud_size) |
| CP.paste_rgba(fr, hud, x + tw - hud[0].shape[1] - 8, y + th - hud[0].shape[0] - 8) |
| yield fr |
| out = f"{O}/sameworld_{name}.mp4" |
| os.makedirs(O, exist_ok=True) |
| CP.write_video(out, frames()) |
| print(f"→ {out}", flush=True) |
| sheet(cases, [f"{name} play {c['k'] + 1}" for c in clips], f"{O}/sheets/sameworld_{name}.jpg") |
| cv2.imwrite(f"{O}/sheets/sameworld_{name}_plan.png", cv2.cvtColor(plan, cv2.COLOR_RGB2BGR)) |
|
|
|
|
| def compose_sameworld_looks(name: str, looks=("night", "snow")) -> None: |
| """start 组：行 = 外观（原首帧 / night / snow），列 = 各 clip 的按键 ⇒ 语义（同一世界）× 首帧（外观）× 按键（动作）三者独立。""" |
| S = f"{A}/sameworld" |
| info = json.load(open(f"{S}/maps/{name}/group.json")) |
| rows_ = [None] + [lk for lk in looks] |
| tiles = [] |
| for lk in rows_: |
| for c in info["clips"]: |
| d = f"{S}/roll/{name}__{c['k']}_{c['start']:06d}" + (f"__{lk}" if lk else "") |
| if not os.path.exists(f"{d}/meta.json"): |
| return |
| tiles.append({"kind": "gen", "case": d, "hud": True, |
| "label": f"Play {c['k'] + 1}" + (f" · {lk.capitalize()}" if lk else " · Original")}) |
| tiles_video(f"{O}/sameworld_{name}_looks.mp4", tiles, len(info["clips"]), |
| "One world, one start: appearance x keys", |
| "Rows: first frame (original / edited to night / snow) · columns: different keys · same semantic world") |
| sheet([t["case"] for t in tiles], [t["label"] for t in tiles], f"{O}/sheets/sameworld_{name}_looks.jpg") |
|
|
|
|
| CP_COLORS = [(255, 64, 200), (40, 220, 255), (255, 200, 40), (120, 255, 90), (255, 110, 60), (170, 120, 255)] |
|
|
|
|
| def main(argv=None) -> None: |
| ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) |
| sp = ap.add_subparsers(dest="cmd", required=True) |
| p = sp.add_parser("app") |
| p.add_argument("--only", nargs="*", default=None) |
| q = sp.add_parser("sameworld") |
| q.add_argument("groups", nargs="+") |
| a = ap.parse_args(argv) |
| if a.cmd == "app": |
| compose_app(a) |
| else: |
| for g in a.groups: |
| compose_sameworld(g, a) |
| compose_sameworld_looks(g) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
