"""EP01 SC02 action test; local weights on the owner's existing :8188 service.""" import argparse import importlib.util import json import secrets import sys import time from pathlib import Path from urllib.parse import quote import requests HERE = Path(__file__).resolve().parent ROOT_HERE = HERE BASE = 'http://192.168.10.51:8188' NAME = '2026-09-29_EP01_LOCKER_PHONE_WAN22_5B_V01' SOURCE = HERE / 'first_frame_v01.png' VARIANT = '5b' WIDTH, HEIGHT, FRAMES, FPS = 1024, 576, 145, 24 POSITIVE = ('A continuous hand-drawn anime medium shot, locked-off camera. The man performs a ' 'clear everyday action with visible arm, shoulder and torso movement. He glances at his phone ' 'for a brief moment, then turns his head and shoulders toward the open locker on the left. ' 'He extends his right arm into the locker and places the phone flat onto the empty shelf. ' 'He lets go of the phone and withdraws his empty right hand. Then he reaches for the open ' 'locker door and swings it shut until the door is closed, ending with his hand resting ' 'briefly on the closed metal door. His eyes follow his hand, his weight shifts naturally ' 'toward the locker. Quiet wistful expression, mouth closed. Clearly readable deliberate ' 'physical movement, natural human timing. Preserve the same adult face, short black hair, ' 'blue uniform and reflective strips, crisp cel shadows and fine drawn outlines. ' 'The background and camera remain stationary. A single uninterrupted shot.') NEGATIVE = ('frozen character, static image, slideshow, camera zoom, camera pan, camera shake, ' 'cuts, new character, talking, lip movement, floating phone, duplicate phone, extra arms, ' 'extra fingers, fused fingers, deformed hands, changing face, changing clothes, morphing, ' 'melting locker, flicker, photorealistic, 3d render, subtitles, watermark, blurry') spec = importlib.util.spec_from_file_location('previous_test', HERE.parent / '2026-09-29_泉堂夜景/render_test.py') previous = importlib.util.module_from_spec(spec) spec.loader.exec_module(previous) previous.HERE = HERE get, save = previous.get, previous.save def build(): assert SOURCE.is_file() assert not (HERE / 'job.json').exists(), 'Already submitted; use --collect' queue = get('/queue') assert not queue['queue_running'] and not queue['queue_pending'], 'Server busy' save('server_stats.json', get('/system_stats')) template = ROOT_HERE.parent / ('2026-09-29_泉堂夜景/source_template.json' if VARIANT == '14b' else '2026-09-29_泉堂夜景/V02_5B/source_template.json') save('source_template.json', json.loads(template.read_text(encoding='utf-8'))) schema = get('/object_info') 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']])) seed = secrets.randbelow(2**48) 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_ti2v_5B_fp16.safetensors', weight_dtype='default') node(6, 'ModelSamplingSD3', model=['2', 0], shift=8.0) node(8, 'CLIPLoader', clip_name='umt5_xxl_fp8_e4m3fn_scaled.safetensors', type='wan', device='default') node(9, 'VAELoader', vae_name='wan2.2_vae.safetensors') node(10, 'CLIPTextEncode', clip=['8', 0], text=POSITIVE) node(11, 'CLIPTextEncode', clip=['8', 0], text=NEGATIVE) node(12, 'Wan22ImageToVideoLatent', vae=['9', 0], start_image=['1', 0], width=1024, height=576, length=145, 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) node(15, 'VAEDecode', samples=['14', 0], vae=['9', 0]) node(16, 'VHS_VideoCombine', images=['15', 0], frame_rate=24.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 == '14b': 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(6, 'ModelSamplingSD3', model=['2', 0], shift=5.0) node(7, 'ModelSamplingSD3', model=['3', 0], shift=5.0) node(9, 'VAELoader', vae_name='wan_2.1_vae.safetensors') node(12, 'WanImageToVideo', positive=['10', 0], negative=['11', 0], vae=['9', 0], start_image=['1', 0], width=WIDTH, height=HEIGHT, length=FRAMES, batch_size=1) common = dict(positive=['12', 0], negative=['12', 1], steps=20, cfg=3.5, 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=10, 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=10, end_at_step=20, return_with_leftover_noise='disable', **common) graph['16']['inputs']['frame_rate'] = FPS for nid, n in graph.items(): s = schema[n['class_type']] assert not s.get('api_node'), 'No paid/API nodes' 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 elif isinstance(kind, list) and key != 'image': assert value in kind, (nid, key, value) 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, kind = n['inputs'].get(key), 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':[360,300], 'flags':{}, 'order':len(ui['nodes']), '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,NAME) 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':WIDTH,'height':HEIGHT,'frames':FRAMES,'fps':FPS,'variant':VARIANT,'submitted_at':time.time()} save('job.json',job) print(json.dumps(job,ensure_ascii=False),flush=True) if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument('--collect',action='store_true') parser.add_argument('--take',type=int,choices=[1,2,3,4],default=1) args = parser.parse_args() if args.take == 2: HERE = ROOT_HERE / 'V02_放下手机' HERE.mkdir(exist_ok=True) previous.HERE = HERE NAME = NAME.replace('V01','V02') POSITIVE = ('Locked-off medium shot, one continuous hand-drawn anime scene. ' 'The man lowers the smartphone in his right hand toward the EMPTY HORIZONTAL ' 'SHELF INSIDE the open locker at waist height. He bends his right elbow downward ' 'and extends his forearm into the locker opening, rotates his wrist so the phone ' 'lies horizontally, screen upward. He sets the smartphone flat on the empty shelf, ' 'opens his fingers to release it, and slowly pulls his empty hand back to his side. ' 'The phone remains visibly lying on the shelf inside the locker. His head tilts ' 'down and his eyes follow the phone. His shoulders lean gently forward then relax. ' 'The locker door stays fully open and does not move throughout this shot. ' 'Clear substantial downward and leftward arm movement, natural human gesture. ' 'The same adult man, same face and blue work uniform, mouth closed. Stable crisp ' '2D cel animation, consistent drawn outlines, stationary background and camera.') NEGATIVE += ', closing locker door, touching door, phone pressed to door, phone disappears, phone floating' if args.take in (3,4): VARIANT = '14b' if args.take == 3: WIDTH, HEIGHT, FRAMES, FPS = 768, 432, 49, 16 HERE = ROOT_HERE / 'V03_14B_无LoRA诊断' else: WIDTH, HEIGHT, FRAMES, FPS = 1024, 576, 81, 16 HERE = ROOT_HERE / 'V04_14B_画质对比' HERE.mkdir(exist_ok=True) previous.HERE = HERE NAME = '2026-09-29_EP01_LOCKER_PHONE_WAN22_14B_NO_LORA_V0' + str(args.take) baseline = json.loads((ROOT_HERE / 'V02_放下手机/workflow_api.json').read_text(encoding='utf-8')) POSITIVE, NEGATIVE = baseline['10']['inputs']['text'], baseline['11']['inputs']['text'] if not args.collect: build() previous.collect()