201 lines
12 KiB
Python
201 lines
12 KiB
Python
"""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()
|