"""Publish a GENERATED turn (audio -> generator -> codec decoder) in the E2E viewer. Third leg of the diagnostic ladder, after `render_solo_gt_turn.py` (no codec, no generator) and `render_solo_codec_roundtrip_turn.py` (codec only). Running all three on the SAME clip separates the error budget: GT wrong -> preprocessing / rest-pose retargeting GT fine, round-trip wrong -> codec reconstruction ceiling round-trip fine, gen odd -> the generator itself READ THE NUMBERS CORRECTLY. The codec is an autoencoder, so its output *should* reproduce GT frame by frame and a per-frame angle error is the right metric. The generator is NOT reproducing GT — it invents a plausible body for the audio, so a frame-by-frame diff against GT is meaningless here (a large value is expected and is not a defect). What is comparable is the DISTRIBUTION: how fast the body moves, how much it rests, how far joints travel, how jittery it is. Those are reported. This also serves as the end-to-end check of the 2026-08-05 publish-path repair (template writer / carried translation / schema-derived bone count) — the unit test only proved write_vrma_from_template reproduces a baked clip; only this path proves body_engine actually uses it. `face.json` and `audio.wav` are copied byte-for-byte from the GT turn, so the body is provably the only difference across all three turns. python render_solo_generated_turn.py \ --wav /data/.../raw/test/solo_20260722_174736_seg000123.wav \ --gt-turn GT_17_median_seg123 --id GEN_17_seg123 """ import argparse import json import os import shutil import sys import time import numpy as np HERE = os.path.dirname(os.path.abspath(__file__)) KEMIX = os.path.abspath(os.path.join(HERE, "..")) WH = os.path.join(KEMIX, "motion-blender", "experiments", "wave-hands") sys.path.insert(0, HERE) sys.path.insert(0, os.path.join(KEMIX, "motion-blender", "lib")) DEFAULT_RUN = "solo_body_v1__50b_vel__scratch__r03" DEFAULT_TEMPLATE = ("/data/mocap-convertor/datasets/solo_20260722_body_v1" "/artifacts/templates/kemix.vrma") DEFAULT_NPZ = "/data/mocap-convertor/datasets/solo_20260722_body_v1/processed/npz" FPS = 30 def motion_stats(quats, trans=None): """Distribution-level descriptors — the fair way to compare a generated body.""" q = np.asarray(quats, dtype=np.float64).copy() for i in range(1, len(q)): # 부호 연속화 s = np.sign(np.sum(q[i] * q[i - 1], axis=-1, keepdims=True)) s[s == 0] = 1 q[i] *= s step = np.degrees(2 * np.linalg.norm(np.diff(q, axis=0)[..., :3], axis=-1)) body = step[:, :20] out = { "speed_deg_per_frame": float(body.mean()), "speed_p95": float(np.percentile(body, 95)), "arms_speed": float(step[:, 6:14].mean()), "fingers_speed": float(step[:, 20:].mean()) if step.shape[1] > 20 else 0.0, # 정지 비율: 전신 속도가 클립 중앙값의 30% 아래인 프레임 "rest_frac": float(np.mean(body.mean(1) < 0.30 * np.median(body.mean(1)))), # 떨림: 각가속의 p99 (프레임간 속도 변화) "jerk_p99": float(np.percentile(np.abs(np.diff(body, axis=0)), 99)), "frames": int(len(q)), } if trans is not None and len(trans) > 1: d = np.diff(np.asarray(trans, dtype=np.float64)[:, [0, 2]], axis=0) out["travel_cm"] = float(np.abs(d).sum() * 100) out["net_travel_cm"] = float(np.linalg.norm(trans[-1][[0, 2]] - trans[0][[0, 2]]) * 100) return out def main(): ap = argparse.ArgumentParser() ap.add_argument("--wav", required=True) ap.add_argument("--gt-turn", required=True, help="face.json/audio.wav를 바이트 복사할 GT 턴") ap.add_argument("--id", required=True) ap.add_argument("--emotion", default="neutral") ap.add_argument("--run-id", default=DEFAULT_RUN) # 런 안의 특정 체크포인트를 고를 때. 기본은 generator/best.pth인데 best는 # **마지막 에폭이 아니라 val 최저 에폭**이라 런마다 제각각이다 (실측 2026-08-06: # r03 2899 / r04b 699 / r04c 299). 그대로 A/B에 올리면 학습량이 2배 차이나는 # 것을 "구조 차이"로 오독한다. 에폭을 맞추려면 epoch_*.pth를 직접 지정한다. ap.add_argument("--ckpt", default=None, help="run_root/generator/best.pth 대신 쓸 체크포인트 경로") ap.add_argument("--template", default=DEFAULT_TEMPLATE) ap.add_argument("--npz", default=DEFAULT_NPZ) # 기본 None = 모델 기본값(base는 yaml guidance_scale=2). 종전 기본 3.0은 # base 경로에서 읽히지도 않아 meta.json에 거짓값만 남겼다 (2026-08-06). ap.add_argument("--guidance", type=float, default=None) # 샘플링 스텝. 종전 고정값 8은 BodyEngine 기본이고 yaml 기본은 2다. # 고친 자로 재측정하니(2026-08-07) r04a·r04c는 **많을수록** 좋다 — # r04a W1 0.420(s4) -> 0.383(s8) -> 0.349(s32), 동적범위 2.7 -> 3.3. # 종전에 "적을수록 좋다"고 읽힌 건 steps가 cond_time=1/steps를 같이 바꿔 # 다른 함수를 부르기 때문이었다(d 아티팩트). ap.add_argument("--steps", type=int, default=8) # 학습의 flow loss는 cond_time=0인 표본만 지도하는데(vrm_lsm.py:198) 추론은 # 항상 1/steps를 넣는다(:148). 이 옵션은 학습이 지도한 좌표로 되돌린다. # r04a 실측: 절대정지 0.008 -> 0.015, 동적범위 3.3 -> 3.7. ap.add_argument("--cond-time-zero", action="store_true") ap.add_argument("--gpu", type=int, default=0) # 샘플러 초기 노이즈 시드. VrmLSM.forward의 torch.randn에는 시드가 없어 # 같은 입력에도 매 실행 결과가 달랐다 — 발행물을 A/B로 비교할 수 없었다 # (2026-08-06, guidance 회귀 검증 중 발견). 미지정이면 난수를 뽑아 쓰되 # **반드시 meta.json에 기록**해 사후 재현이 항상 가능하게 한다. ap.add_argument("--seed", type=int, default=None) ap.add_argument("--compare-npz", default=None, help="같은 클립의 GT npz — 분포 비교용(오차 아님)") ap.add_argument("--force", action="store_true") args = ap.parse_args() gt_dir = os.path.join(HERE, "outputs", "turns", args.gt_turn) for name in ("face.json", "audio.wav"): if not os.path.isfile(os.path.join(gt_dir, name)): raise FileNotFoundError(f"{gt_dir}/{name} — GT 턴을 먼저 발행하라") turn_dir = os.path.join(HERE, "outputs", "turns", args.id) if os.path.exists(turn_dir) and not args.force: raise FileExistsError(f"{turn_dir} 이미 있음; --force 또는 다른 id") run_root = os.path.join(WH, "outputs", "gen_train", "runs", args.run_id) ckpt = args.ckpt or os.path.join(run_root, "generator", "best.pth") selection = os.path.join(run_root, "codec_selection.json") for p in (ckpt, selection, args.template): if not os.path.isfile(p): raise FileNotFoundError(p) # 런의 구조 스위치(사이클 2/2, 2026-08-06). 체크포인트만으로는 부족하다 — # audio_rope는 rope_emb.inv_freq 버퍼로 탐지되지만 time_major_tokens는 # 파라미터가 없어 흔적이 남지 않는다. variant.json(prepare_r04_runs.py가 기록)을 # 우선 읽는다. 틀리면 로드가 깨지거나 — 더 나쁘게는 조용히 **다른 모델을 발행**한다. # gate_audio_conditioning.py:405-432와 같은 규칙이다. import torch as _t _ck = _t.load(ckpt, map_location="cpu", weights_only=False) _sd = _ck.get("model_state_dict", _ck) ckpt_epoch = _ck.get("epoch") dparams = dict(time_major_tokens=False, audio_rope=False) vpath = os.path.join(run_root, "variant.json") if os.path.isfile(vpath): ta = json.load(open(vpath)).get("train_args", []) dparams["time_major_tokens"] = "--time-major-tokens" in ta dparams["audio_rope"] = "--audio-rope" in ta print(f"구조: variant.json — {dparams}") elif any("rope_emb" in k for k in _sd): raise SystemExit( f"{ckpt}: rope 체크포인트인데 variant.json이 없다. " "time_major_tokens 여부를 알 수 없어 조용히 틀린 모델을 발행할 위험이 있다") else: print("구조: 기본(off) — variant.json 없음, 체크포인트에 rope 흔적 없음") del _ck, _sd print(f"체크포인트: epoch {ckpt_epoch} — {os.path.basename(ckpt)}") started = time.time() from body_engine import BodyEngine # 얼굴 패키지 임포트 금지 eng = BodyEngine(ckpt=ckpt, vqvae_dir=run_root, meanstd_dir=os.path.join(run_root, "mean_std"), npz=args.npz, fingers=True, steps=args.steps, device=f"cuda:{args.gpu}", model_kind="base", denoiser_params=dparams, codec_selection=selection, continuous_decode=True, template_vrma=args.template) import torch # 시드 고정용 (엔진 로드 후) if args.cond_time_zero: _cfg = eng.model.apply_classifier_free_guidance def _zero_d(*aa, **kw): if kw.get("cond_time") is not None: kw["cond_time"] = torch.zeros_like(kw["cond_time"]) return _cfg(*aa, **kw) eng.model.apply_classifier_free_guidance = _zero_d seed = args.seed if args.seed is not None else int.from_bytes(os.urandom(4), "little") torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) np.random.seed(seed) t0 = time.time() quats, frames, root = eng.generate_take(args.wav, args.emotion, guidance=args.guidance) gen_s = time.time() - t0 os.makedirs(turn_dir, exist_ok=True) report = eng.write_vrma(quats, os.path.join(turn_dir, "body.vrma"), root_translation=root) for name in ("face.json", "audio.wav"): shutil.copyfile(os.path.join(gt_dir, name), os.path.join(turn_dir, name)) face_frames = json.load(open(os.path.join(turn_dir, "face.json")))["num_frames"] stats = {"generated": motion_stats(quats, root)} if args.compare_npz: with np.load(args.compare_npz, allow_pickle=False) as d: stats["ground_truth"] = motion_stats(d["vrm_rotations"].astype(np.float64), d["trans"].astype(np.float64)) meta = { "id": args.id, "emotion": args.emotion, "fps": FPS, "face_frames": int(face_frames), "body_frames": int(frames), "face": "face.json", "body": "body.vrma", "audio": "audio.wav", "body_source": "generator + codec decoder (continuous)", "body_writer": "baked_template_channel_write", "bones": len(eng.bone_names), "run_id": args.run_id, "generator_ckpt": ckpt, # 어느 에폭인지 반드시 남긴다. best.pth는 런마다 다른 에폭이라 이게 없으면 # 학습량이 다른 것들을 나란히 놓고 "구조 차이"로 오독하게 된다 (2026-08-06 실측). "generator_epoch": ckpt_epoch, "guidance": eng.effective_guidance(args.guidance), # 요청값 아닌 실효값 "steps": int(args.steps), "cond_time_zero": bool(args.cond_time_zero), "rng_seed": int(seed), # 이 턴을 재현하는 값 "compare_with": args.gt_turn, "root_translation": "codec velocity integrated from origin", "motion_stats": stats, "writer_report": {k: v for k, v in (report or {}).items() if k != "written"}, "note": ("생성기는 GT를 재현하지 않는다 — 오디오에 맞는 새 모션을 만든다. " "프레임별 각도 차이는 의미가 없고 분포(속도·정지비율·이동)를 본다."), "timing": {"face_s": 0.0, "body_s": round(gen_s, 2), "total_s": round(time.time() - started, 2)}, } with open(os.path.join(turn_dir, "meta.json"), "w") as f: json.dump(meta, f, ensure_ascii=False, indent=1) print(f"{args.id}: {frames}프레임, 생성 {gen_s:.2f}s, 본 {len(eng.bone_names)}, " f"guidance {meta['guidance']}, seed {seed}") if report: print(f" writer: 기록 {len(report['written'])}본, 건너뜀 {report['skipped']}, " f"이동 실림={report['root_translation']}") for k, v in stats.items(): print(f" [{k}] " + " ".join( f"{n} {x:.3f}" if isinstance(x, float) else f"{n} {x}" for n, x in v.items())) print(turn_dir) if __name__ == "__main__": main()