"""One-shot Wan 2.2 test on the owner's existing ComfyUI. No server installs.""" import argparse import copy import json import secrets import sys import time from pathlib import Path from urllib.parse import quote import requests BASE = 'http://192.168.10.51:8188' HERE = Path(__file__).resolve().parent ROOT = HERE.parents[2] SOURCE = ROOT / '03_分镜与镜头/S01E01_V05_制作包/06_参考图/BG-EP01-CHAPEL-NIGHT-V01-REVIEW.png' TEMPLATE = 'workflows/production_pack_20260904/03_generation_cinematic/09_wan22_i2v.json' NAME = '2026-09-29_EP01_CHAPEL_NIGHT_WAN22_TEST_V01' POSITIVE = ('A single uninterrupted wide establishing shot. The camera makes a very slow, ' 'gentle dolly forward down the central aisle, keeping the stone statue centered. ' 'The small candle flames and the distant hearth fire flicker softly, casting faint ' 'warm moving light on the stone. The statue is inanimate stone and stays completely ' 'rigid, with an unchanged face and hands. Wooden pews, masonry and hanging cloth remain ' 'stable. Preserve the reference image composition, delicate drawn contours, richly ' 'hand-painted anime background, muted colors, cool blue window light and warm amber ' 'candlelight. Restrained movement, quiet nighttime atmosphere, stable exposure.') NEGATIVE = ('people, living statue, blinking statue, moving statue, speaking, supernatural glow, ' 'magic particles, new objects, flickering geometry, morphing, melting, warped architecture, ' 'camera shake, fast zoom, orbit, cuts, oversaturated, overexposed, photorealistic, ' '3d render, text, subtitles, watermark, blurry, low quality') def get(path): r = requests.get(BASE + path, timeout=30) r.raise_for_status() return r.json() def save(name, obj): (HERE / name).write_text(json.dumps(obj, ensure_ascii=False, indent=2), encoding='utf-8') def build(): assert SOURCE.is_file(), SOURCE if (HERE / 'job.json').exists(): raise RuntimeError('Job already exists. Use --collect; do not accidentally submit twice.') q = get('/queue') if q['queue_running'] or q['queue_pending']: raise RuntimeError('Server is busy; no job submitted.') stats = get('/system_stats') save('server_stats.json', stats) if VARIANT == '5b': r = requests.get('https://raw.githubusercontent.com/Comfy-Org/workflow_templates/main/templates/video_wan2_2_5B_ti2v.json', timeout=30) r.raise_for_status() source_template = r.json() else: source_template = get('/userdata/' + quote(TEMPLATE, safe='')) save('source_template.json', source_template) schema = get('/object_info') seed = secrets.randbelow(2**48) with SOURCE.open('rb') as f: r = requests.post(BASE + '/upload/image', files={'image': (NAME + '.png', f, 'image/png')}, data={'type': 'input', 'subfolder': 'hearthfire_tests', 'overwrite': 'false'}, timeout=60) r.raise_for_status() uploaded = r.json() input_name = '/'.join(filter(None, [uploaded.get('subfolder'), uploaded['name']])) graph = {} def node(i, cls, **inputs): graph[str(i)] = {'class_type': cls, 'inputs': inputs} node(1, 'LoadImage', image=input_name) node(2, 'UNETLoader', unet_name='wan2.2_i2v_high_noise_14B_fp8_scaled.safetensors', weight_dtype='default') node(3, 'UNETLoader', unet_name='wan2.2_i2v_low_noise_14B_fp8_scaled.safetensors', weight_dtype='default') node(4, 'LoraLoaderModelOnly', model=['2', 0], lora_name='wan2.2_i2v_lightx2v_4steps_lora_v1_high_noise.safetensors', strength_model=1.0) node(5, 'LoraLoaderModelOnly', model=['3', 0], lora_name='wan2.2_i2v_lightx2v_4steps_lora_v1_low_noise.safetensors', strength_model=1.0) node(6, 'ModelSamplingSD3', model=['4', 0], shift=5.0) node(7, 'ModelSamplingSD3', model=['5', 0], shift=5.0) node(8, 'CLIPLoader', clip_name='umt5_xxl_fp8_e4m3fn_scaled.safetensors', type='wan', device='default') node(9, 'VAELoader', vae_name='wan_2.1_vae.safetensors') node(10, 'CLIPTextEncode', clip=['8', 0], text=POSITIVE) node(11, 'CLIPTextEncode', clip=['8', 0], text=NEGATIVE) node(12, 'WanImageToVideo', positive=['10', 0], negative=['11', 0], vae=['9', 0], start_image=['1', 0], width=768, height=432, length=81, batch_size=1) common = dict(positive=['12', 0], negative=['12', 1], steps=4, cfg=1.0, sampler_name='euler', scheduler='simple') node(13, 'KSamplerAdvanced', model=['6', 0], latent_image=['12', 2], add_noise='enable', noise_seed=seed, start_at_step=0, end_at_step=2, return_with_leftover_noise='enable', **common) node(14, 'KSamplerAdvanced', model=['7', 0], latent_image=['13', 0], add_noise='disable', noise_seed=0, start_at_step=2, end_at_step=4, return_with_leftover_noise='disable', **common) node(15, 'VAEDecode', samples=['14', 0], vae=['9', 0]) node(16, 'VHS_VideoCombine', images=['15', 0], frame_rate=16.0, loop_count=0, filename_prefix='hearthfire_tests/' + NAME, format='video/h264-mp4', pingpong=False, save_output=True, pix_fmt='yuv420p', crf=18, save_metadata=True, trim_to_audio=False) if VARIANT == '5b': graph = {k: v for k, v in graph.items() if k in ('1', '8', '10', '11', '15', '16')} node(2, 'UNETLoader', unet_name='wan2.2_ti2v_5B_fp16.safetensors', weight_dtype='default') node(6, 'ModelSamplingSD3', model=['2', 0], shift=8.0) node(9, 'VAELoader', vae_name='wan2.2_vae.safetensors') node(12, 'Wan22ImageToVideoLatent', vae=['9', 0], start_image=['1', 0], width=768, height=448, length=121, batch_size=1) node(14, 'KSampler', model=['6', 0], positive=['10', 0], negative=['11', 0], latent_image=['12', 0], seed=seed, steps=20, cfg=5.0, sampler_name='uni_pc', scheduler='simple', denoise=1.0) graph['16']['inputs']['frame_rate'] = 24.0 # Live type, required-input, model-name and connection validation. for nid, n in graph.items(): s = schema[n['class_type']] assert not s.get('api_node'), 'Paid/API nodes are out of scope' specs = {**s['input'].get('required', {}), **s['input'].get('optional', {})} assert set(s['input'].get('required', {})) <= n['inputs'].keys() for key, value in n['inputs'].items(): if key not in specs: assert n['class_type'] == 'VHS_VideoCombine' continue kind = specs[key][0] if isinstance(value, list): origin, slot = value assert schema[graph[origin]['class_type']]['output'][slot] == kind, (nid, key) elif isinstance(kind, list) and key != 'image': assert value in kind, (nid, key, value) # Emit a flat GUI graph from the exact API graph, retaining widget order. ui = {'last_node_id': 16, 'last_link_id': 0, 'nodes': [], 'links': [], 'groups': [], 'config': {}, 'extra': {'ds': {'scale': 0.4, 'offset': [0, 0]}}, 'version': 0.4} for nid, n in graph.items(): s = schema[n['class_type']] slots, widgets = [], [] specs = {**s['input'].get('required', {}), **s['input'].get('optional', {})} for key, spec in specs.items(): val = n['inputs'].get(key) kind = spec[0] if isinstance(val, list) or (isinstance(kind, str) and kind not in ('INT', 'FLOAT', 'BOOLEAN', 'STRING', 'COMBO')): slots.append({'name': key, 'type': kind, 'link': None}) elif key in n['inputs']: widgets.append(val) if len(spec) > 1 and spec[1].get('control_after_generate'): widgets.append('fixed') if n['class_type'] == 'LoadImage': widgets.append('image') if n['class_type'] == 'VHS_VideoCombine': widgets = {k: v for k, v in n['inputs'].items() if not isinstance(v, list)} ui['nodes'].append({'id': int(nid), 'type': n['class_type'], 'pos': [0, 0], 'size': [340, 260], 'flags': {}, 'order': int(nid)-1, 'mode': 0, 'inputs': slots, 'outputs': [{'name': name, 'type': typ, 'links': []} for name, typ in zip(s['output_name'], s['output'])], 'properties': {'Node name for S&R': n['class_type']}, 'widgets_values': widgets}) indexed = {str(n['id']): n for n in ui['nodes']} for nid, n in graph.items(): for key, val in n['inputs'].items(): if not isinstance(val, list): continue origin, oslot = val slot = next(i for i, item in enumerate(indexed[nid]['inputs']) if item['name'] == key) ui['last_link_id'] += 1 link = ui['last_link_id'] kind = indexed[origin]['outputs'][oslot]['type'] ui['links'].append([link, int(origin), oslot, int(nid), slot, kind]) indexed[nid]['inputs'][slot]['link'] = link indexed[origin]['outputs'][oslot]['links'].append(link) sys.path.insert(0, 'C:/Users/zhang/.codex/skills/comfyui') from workflow_layout import auto_layout, inspect, fit_group ui = auto_layout(ui) fit_group(ui, 'EP01 chapel - Wan 2.2 - 5s visual test') layout = inspect(ui) save('layout_check.json', layout) print('Layout:', layout['summary'], flush=True) save('workflow_api.json', graph) save('workflow_gui.json', ui) target = 'workflows/hearthfire_tests/' + NAME + '.json' r = requests.post(BASE + '/userdata/' + quote(target, safe='') + '?overwrite=false', json=ui, timeout=30) r.raise_for_status() assert get('/userdata/' + quote(target, safe='')) == ui r = requests.post(BASE + '/prompt', json={'prompt': graph, 'client_id': NAME, 'extra_data': {'extra_pnginfo': {'workflow': ui}}}, timeout=30) save('submission.json', r.json()) r.raise_for_status() job = {**r.json(), 'seed': seed, 'workflow': target, 'source': str(SOURCE), 'input': uploaded, 'width': 768, 'height': 448 if VARIANT == '5b' else 432, 'frames': 121 if VARIANT == '5b' else 81, 'fps': 24 if VARIANT == '5b' else 16, 'variant': VARIANT, 'submitted_at': time.time()} save('job.json', job) print(json.dumps(job, ensure_ascii=False), flush=True) def collect(): job = json.loads((HERE / 'job.json').read_text(encoding='utf-8')) pid = job['prompt_id'] for attempt in range(160): history = get('/history/' + pid).get(pid) if history: save('history.json', history) if history['status']['status_str'] != 'success': errors = [data for kind, data in history['status']['messages'] if kind == 'execution_error'] print(json.dumps([{k: e.get(k) for k in ('node_id', 'node_type', 'exception_type', 'exception_message')} for e in errors], ensure_ascii=False), flush=True) raise RuntimeError('Generation failed; inspect history.json') outputs = [] for result in history.get('outputs', {}).values(): for entry in result.get('gifs', []) + result.get('videos', []) + result.get('images', []): if not entry.get('filename', '').endswith('.mp4'): continue r = requests.get(BASE + '/view', params={k: entry[k] for k in ('filename', 'subfolder', 'type')}, timeout=90) r.raise_for_status() dest = HERE / entry['filename'] dest.write_bytes(r.content) outputs.append(str(dest)) assert outputs, 'Successful job has no MP4 output' save('outputs.json', outputs) print(json.dumps({'outputs': outputs, 'elapsed_s': round(time.time()-job['submitted_at'])}, ensure_ascii=False), flush=True) return if attempt % 2 == 0: print('Waiting for job', pid, 'elapsed', round(time.time()-job['submitted_at']), 'seconds', flush=True) time.sleep(15) raise TimeoutError('Job retained on server; run --collect to resume waiting.') if __name__ == '__main__': p = argparse.ArgumentParser() p.add_argument('--collect', action='store_true') p.add_argument('--variant', choices=['14b', '5b'], default='14b') args = p.parse_args() VARIANT = args.variant if VARIANT == '5b': HERE = HERE / 'V02_5B' HERE.mkdir(exist_ok=True) NAME = NAME.replace('V01', 'V02_5B') if not args.collect: build() collect()