mirror of
https://github.com/vladmandic/automatic
synced 2026-09-19 01:04:32 +02:00
Merge pull request #5041 from vladmandic/feat/ltx-keyword-core
feat(ltx): add a keyword core to the ltx runner
This commit is contained in:
Vladimir Mandic
GitHub
+387
-308
@@ -11,9 +11,9 @@ from modules.ltx.ltx_util import get_bucket, get_frames, load_model, load_upsamp
|
||||
|
||||
apply_ltx_diffusers_patch()
|
||||
from modules.processing_callbacks import diffusers_callback
|
||||
from modules.video_models import video_run, video_utils
|
||||
from modules.video_models.video_vae import set_vae_params
|
||||
from modules.video_models.video_save import save_video, get_audio_rate
|
||||
from modules.video_models.video_utils import check_av
|
||||
|
||||
|
||||
debug = log.trace if os.environ.get('SD_VIDEO_DEBUG', None) is not None else lambda *args, **kwargs: None
|
||||
@@ -22,6 +22,7 @@ upsample_pipe = None
|
||||
upsample_pipe_2x = None
|
||||
|
||||
STAGE2_DEV_LORA_ADAPTER = 'ltx2_stage2_distilled'
|
||||
I2V_IMAGE_CLASSES = ('LTXImageToVideoPipeline', 'LTX2ImageToVideoPipeline') # the i2v pipes that take the init image as a plain kwarg rather than as a condition
|
||||
|
||||
|
||||
def _prompt_tensors_to_device(*tensors):
|
||||
@@ -103,7 +104,7 @@ def _latent_pass(caps, prompt_embeds, prompt_attention_mask, negative_prompt_emb
|
||||
base_args['image_cond_noise_scale'] = image_cond_noise_scale
|
||||
if caps.supports_multi_condition and conditions:
|
||||
base_args['conditions'] = conditions
|
||||
if caps.is_i2v and caps.repo_cls_name in ('LTXImageToVideoPipeline', 'LTX2ImageToVideoPipeline') and image is not None:
|
||||
if caps.is_i2v and caps.repo_cls_name in I2V_IMAGE_CLASSES and image is not None:
|
||||
base_args['image'] = image
|
||||
if caps.family == '2.x' and caps.is_distilled:
|
||||
from diffusers.pipelines.ltx2.utils import DISTILLED_SIGMA_VALUES
|
||||
@@ -118,88 +119,92 @@ def _latent_pass(caps, prompt_embeds, prompt_attention_mask, negative_prompt_emb
|
||||
return latents
|
||||
|
||||
|
||||
def run_ltx(task_id,
|
||||
_ui_state,
|
||||
model: str,
|
||||
prompt: str,
|
||||
negative: str,
|
||||
styles: list,
|
||||
width: int,
|
||||
height: int,
|
||||
frames: int,
|
||||
auto_duration: bool,
|
||||
steps: int,
|
||||
sampler_index: int,
|
||||
guidance_scale: float,
|
||||
sampler_shift: float,
|
||||
dynamic_shift: bool,
|
||||
seed: int,
|
||||
upsample_enable: bool,
|
||||
upsample_ratio: float,
|
||||
refine_enable: bool,
|
||||
refine_strength: float,
|
||||
condition_strength: float,
|
||||
ltx_init_image,
|
||||
condition_last,
|
||||
condition_files,
|
||||
condition_video,
|
||||
condition_video_frames: int,
|
||||
condition_video_skip: int,
|
||||
decode_timestep: float,
|
||||
image_cond_noise_scale: float,
|
||||
mp4_fps: int,
|
||||
mp4_interpolate: int,
|
||||
mp4_codec: str,
|
||||
mp4_ext: str,
|
||||
mp4_opt: str,
|
||||
mp4_video: bool,
|
||||
mp4_frames: bool,
|
||||
mp4_sf: bool,
|
||||
mp4_thumb: bool,
|
||||
audio_enable: bool,
|
||||
_overrides,
|
||||
*args,
|
||||
**_kwargs,
|
||||
):
|
||||
def reject(msg: str, code: int):
|
||||
"""Refuse the run with a logged reason. The message otherwise only travels in the raised error,
|
||||
and a caller that turns it into a string leaves no trace of why the job did nothing."""
|
||||
if code >= 500:
|
||||
log.error(f'Video: op=ltx code={code} {msg}')
|
||||
else:
|
||||
log.info(f'Video: op=ltx code={code} {msg}')
|
||||
raise video_run.VideoError(msg, code)
|
||||
|
||||
def abort(e, ok: bool = False, p=None):
|
||||
if ok:
|
||||
log.info(e)
|
||||
else:
|
||||
log.error(f'Video: cls={shared.sd_model.__class__.__name__} op=base {e}')
|
||||
errors.display(e, 'LTX')
|
||||
if p is not None:
|
||||
extra_networks.deactivate(p)
|
||||
shared.state.end()
|
||||
progress.finish_task(task_id)
|
||||
yield None, f'LTX Error: {str(e)}'
|
||||
|
||||
def run(model: str, *,
|
||||
prompt: str,
|
||||
negative: str = '',
|
||||
styles: list | None = None,
|
||||
width: int = 768,
|
||||
height: int = 512,
|
||||
frames: int = 121,
|
||||
auto_duration: bool = False,
|
||||
steps: int = 0, # <=0 takes the model default
|
||||
sampler_name: str = 'Default',
|
||||
sampler_shift: float = -1.0, # <0 keeps the model default
|
||||
dynamic_shift: bool = False,
|
||||
seed: int = -1,
|
||||
guidance_scale: float = -1.0, # <=0 takes the model default
|
||||
upsample_enable: bool = False,
|
||||
upsample_ratio: float = 2.0,
|
||||
refine_enable: bool = False,
|
||||
refine_strength: float = 0.4,
|
||||
condition_strength: float = 1.0,
|
||||
init_image=None,
|
||||
condition_last=None,
|
||||
condition_files: list | None = None,
|
||||
condition_video: str | None = None,
|
||||
condition_video_frames: int = -1,
|
||||
condition_video_skip: int = 0,
|
||||
decode_timestep: float = 0.05,
|
||||
image_cond_noise_scale: float = 0.025,
|
||||
audio: bool = True,
|
||||
mp4_fps: int = 24,
|
||||
mp4_interpolate: int = 0,
|
||||
mp4_codec: str = 'libx264',
|
||||
mp4_ext: str = 'mp4',
|
||||
mp4_opt: str = 'crf=16',
|
||||
mp4_video: bool = True,
|
||||
mp4_frames: bool = False,
|
||||
mp4_sf: bool = False,
|
||||
mp4_thumb: bool = True,
|
||||
override_settings=None,
|
||||
ui_state=None,
|
||||
scripts=None,
|
||||
script_args=(),
|
||||
per_script_args: dict | None = None,
|
||||
extra_p: dict | None = None,
|
||||
) -> video_run.VideoResult:
|
||||
"""Generate one LTX video and save it.
|
||||
|
||||
Every failure leaves as a VideoError whose code follows HTTP semantics, with 499 reserved for
|
||||
an interrupt so a cancel is distinguishable from a crash. LTX decodes through its own VAE path,
|
||||
so it takes no vae_type: the decode is always full.
|
||||
"""
|
||||
if model is None or len(model) == 0 or model == 'None':
|
||||
yield from abort('Video: no model selected', ok=True)
|
||||
return
|
||||
reject('no model selected', 400)
|
||||
if model.startswith('─'):
|
||||
yield from abort('Video: dropdown separator selected, pick an actual model below', ok=True)
|
||||
return
|
||||
check_av()
|
||||
progress.add_task_to_queue(task_id)
|
||||
reject('dropdown separator selected, pick an actual model below', 400)
|
||||
if mp4_video and video_utils.check_av() is None:
|
||||
reject('video encoding is unavailable: the av package failed to load', 500)
|
||||
|
||||
with call_queue.get_lock():
|
||||
progress.start_task(task_id)
|
||||
memstats.reset_stats()
|
||||
timer.process.reset()
|
||||
yield None, 'LTX: Loading...'
|
||||
engine = 'LTX Video'
|
||||
load_model(engine, model)
|
||||
caps = ltx_capabilities.get_caps(model)
|
||||
cls = shared.sd_model.__class__.__name__ if shared.sd_loaded else None
|
||||
if caps is None or cls is None or not cls.startswith('LTX'):
|
||||
reject(f'selected model is not LTX: model="{model}" cls={cls}', 400)
|
||||
takes_init_image = caps.is_i2v and caps.repo_cls_name in I2V_IMAGE_CLASSES
|
||||
if takes_init_image and init_image is None:
|
||||
reject('No input image provided. Please upload or select an image.', 400)
|
||||
|
||||
engine = 'LTX Video'
|
||||
load_model(engine, model)
|
||||
caps = ltx_capabilities.get_caps(model)
|
||||
if caps is None or not shared.sd_model.__class__.__name__.startswith('LTX'):
|
||||
yield from abort(f'Video: cls={shared.sd_model.__class__.__name__} selected model is not LTX', ok=True)
|
||||
return
|
||||
|
||||
auto_frames = bool(auto_duration) and caps.supports_auto_duration
|
||||
if auto_duration and not auto_frames:
|
||||
log.warning(f'LTX: model="{model}" auto duration unsupported, using frames={get_frames(frames)}')
|
||||
steps = int(steps) if steps is not None and int(steps) > 0 else caps.default_steps
|
||||
cfg_scale = float(guidance_scale) if guidance_scale is not None and guidance_scale > 0 else caps.default_cfg
|
||||
auto_frames = bool(auto_duration) and caps.supports_auto_duration
|
||||
if auto_duration and not auto_frames:
|
||||
log.warning(f'LTX: model="{model}" auto duration unsupported, using frames={get_frames(frames)}')
|
||||
|
||||
p = None
|
||||
t0 = time.time()
|
||||
try:
|
||||
# Lightricks TI2VidTwoStagesPipeline: Stage 1 at half-res, 2x upsample, Stage 2 refine at target.
|
||||
# Auto-couple when the user picks Refine but not Upsample. Both Dev and Distilled refine paths
|
||||
# expect upsampled latents; same-res refine on Distilled produces oversaturation. Condition
|
||||
@@ -234,19 +239,10 @@ def run_ltx(task_id,
|
||||
final_h = target_h
|
||||
log.debug(f'LTX: resolution planning target={target_w}x{target_h} base={base_w}x{base_h} final={final_w}x{final_h} upsample={auto_refine_upsample}')
|
||||
|
||||
videojob = shared.state.begin('Video', task_id=task_id)
|
||||
shared.state.job_count = 1
|
||||
|
||||
from modules.video_models import models_def, video_overrides
|
||||
selected = next((m for m in models_def.models.get(engine, []) if m.name == model), None)
|
||||
selected = models_def.find(engine, model)
|
||||
|
||||
if caps.is_i2v and caps.repo_cls_name in ('LTXImageToVideoPipeline', 'LTX2ImageToVideoPipeline') and ltx_init_image is None:
|
||||
yield from abort('No input image provided. Please upload or select an image.', ok=True)
|
||||
return
|
||||
|
||||
condition_images = []
|
||||
if ltx_init_image is not None:
|
||||
condition_images.append(ltx_init_image)
|
||||
condition_images = [init_image] if init_image is not None else []
|
||||
conditions = []
|
||||
conditions_stage2 = []
|
||||
if caps.supports_multi_condition:
|
||||
@@ -269,9 +265,8 @@ def run_ltx(task_id,
|
||||
else:
|
||||
conditions_stage2 = conditions
|
||||
|
||||
sampler_name = processing.get_sampler_name(sampler_index)
|
||||
sd_samplers.create_sampler(sampler_name, shared.sd_model)
|
||||
log.debug(f'Video: cls={shared.sd_model.__class__.__name__} op=init caps={caps.family} styles={styles} sampler={shared.sd_model.scheduler.__class__.__name__}')
|
||||
log.debug(f'Video: cls={cls} op=init caps={caps.family} styles={styles} sampler={shared.sd_model.scheduler.__class__.__name__}')
|
||||
|
||||
from modules.paths import resolve_output_path
|
||||
p = processing.StableDiffusionProcessingVideo(
|
||||
@@ -280,29 +275,35 @@ def run_ltx(task_id,
|
||||
video_model=model,
|
||||
prompt=prompt,
|
||||
negative_prompt=negative,
|
||||
styles=styles,
|
||||
styles=styles or [],
|
||||
seed=int(seed) if seed is not None else -1,
|
||||
sampler_name=sampler_name,
|
||||
sampler_shift=float(sampler_shift),
|
||||
steps=int(steps),
|
||||
steps=steps,
|
||||
width=base_w,
|
||||
height=base_h,
|
||||
frames=get_frames(frames),
|
||||
cfg_scale=float(guidance_scale) if guidance_scale is not None and guidance_scale > 0 else caps.default_cfg,
|
||||
cfg_scale=cfg_scale,
|
||||
denoising_strength=float(condition_strength) if condition_strength is not None else 1.0,
|
||||
init_image=ltx_init_image,
|
||||
init_image=init_image,
|
||||
vae_type='Default',
|
||||
vae_tile_frames=16,
|
||||
override_settings=video_run.normalize_override_settings(override_settings),
|
||||
)
|
||||
processing.fix_seed(p)
|
||||
p.state = ui_state
|
||||
p.do_not_save_grid = True
|
||||
p.do_not_save_samples = not mp4_frames
|
||||
p.outpath_samples = resolve_output_path(shared.opts.outdir_samples, shared.opts.outdir_video)
|
||||
p.ops.append('video')
|
||||
if per_script_args:
|
||||
p.per_script_args.update(per_script_args)
|
||||
for k, v in (extra_p or {}).items():
|
||||
setattr(p, k, v)
|
||||
|
||||
p.scripts = scripts_manager.scripts_video
|
||||
p.script_args = args
|
||||
processed: processing.Processed = scripts_manager.scripts_video.run(p, *args)
|
||||
p.scripts = scripts if scripts is not None else scripts_manager.scripts_video
|
||||
p.script_args = tuple(script_args)
|
||||
p.scripts.run(p, *p.script_args)
|
||||
|
||||
p.task_args['num_inference_steps'] = p.steps
|
||||
p.task_args['width'] = p.width
|
||||
@@ -318,9 +319,9 @@ def run_ltx(task_id,
|
||||
if caps.supports_multi_condition and conditions:
|
||||
p.task_args['conditions'] = conditions
|
||||
|
||||
if caps.is_i2v and caps.repo_cls_name in ('LTXImageToVideoPipeline', 'LTX2ImageToVideoPipeline') and ltx_init_image is not None:
|
||||
if takes_init_image:
|
||||
from modules import images
|
||||
p.task_args['image'] = images.resize_image(resize_mode=2, im=ltx_init_image, width=p.width, height=p.height, upscaler_name=None, output_type='pil')
|
||||
p.task_args['image'] = images.resize_image(resize_mode=2, im=init_image, width=p.width, height=p.height, upscaler_name=None, output_type='pil')
|
||||
|
||||
if caps.family == '2.x' and caps.is_distilled:
|
||||
from diffusers.pipelines.ltx2.utils import DISTILLED_SIGMA_VALUES
|
||||
@@ -341,19 +342,18 @@ def run_ltx(task_id,
|
||||
if selected is not None:
|
||||
video_overrides.set_overrides(p, selected)
|
||||
|
||||
t0 = time.time()
|
||||
t_offload = time.time()
|
||||
shared.sd_model = sd_models.apply_balanced_offload(shared.sd_model, silent=True)
|
||||
t1 = time.time()
|
||||
timer.process.add('offload', t1 - t_offload)
|
||||
|
||||
samplejob = shared.state.begin('Sample')
|
||||
yield None, 'LTX: Generate in progress...'
|
||||
|
||||
audio = None
|
||||
audio_out = None
|
||||
pixels = None
|
||||
frames_out = None
|
||||
latents = None
|
||||
prompt_embeds = prompt_attention_mask = negative_prompt_embeds = negative_prompt_attention_mask = None
|
||||
needs_latent_path = upsample_enable or refine_enable
|
||||
|
||||
try:
|
||||
with video_utils.phase('Sample'):
|
||||
if needs_latent_path:
|
||||
if p.scripts is not None and isinstance(p.scripts, scripts_manager.ScriptRunner):
|
||||
p.scripts.before_process(p)
|
||||
@@ -401,45 +401,40 @@ def run_ltx(task_id,
|
||||
else:
|
||||
processed = processing.process_images(p)
|
||||
if processed is None or processed.images is None or len(processed.images) == 0:
|
||||
yield from abort('Video: process_images returned no frames', ok=True, p=p)
|
||||
return
|
||||
# process_images swallows the interrupt assertion, so an empty result is the
|
||||
# only place a cancel and a genuine failure are still distinguishable
|
||||
if shared.state.interrupted or shared.state.skipped:
|
||||
reject('interrupted', 499)
|
||||
reject('process_images returned no frames', 500)
|
||||
pixels = processed.images
|
||||
raw_audio = getattr(processed, 'audio', None)
|
||||
if raw_audio is not None:
|
||||
# Strip batch dim from (B, 2, N); write_audio expects (2, N) for the
|
||||
# transpose-to-interleaved path used by AAC s16.
|
||||
audio = raw_audio[0].float().cpu() if raw_audio.ndim == 3 else raw_audio.float().cpu()
|
||||
latents = None
|
||||
except AssertionError as e:
|
||||
yield from abort(e, ok=True, p=p)
|
||||
return
|
||||
except Exception as e:
|
||||
yield from abort(e, ok=False, p=p)
|
||||
return
|
||||
audio_out = raw_audio[0].float().cpu() if raw_audio.ndim == 3 else raw_audio.float().cpu()
|
||||
|
||||
t2 = time.time()
|
||||
# silent=True everywhere in run_ltx: per-module stats were already dumped during the
|
||||
# load-time balanced_offload pass. Upsample/refine boundaries force a rebuild because
|
||||
# the global offload_hook_instance is keyed on checkpoint_name (sd_offload.py:488),
|
||||
# but re-logging the same inventory adds noise without information.
|
||||
shared.sd_model = sd_models.apply_balanced_offload(shared.sd_model, silent=True)
|
||||
devices.torch_gc(force=True, reason='ltx:base')
|
||||
t3 = time.time()
|
||||
timer.process.add('offload', t1 - t0)
|
||||
timer.process.add('base', t2 - t1)
|
||||
timer.process.add('offload', t3 - t2)
|
||||
shared.state.end(samplejob)
|
||||
t2 = time.time()
|
||||
# silent=True everywhere: per-module stats were already dumped during the load-time
|
||||
# balanced_offload pass. Upsample/refine boundaries force a rebuild because the global
|
||||
# offload_hook_instance is keyed on checkpoint_name (sd_offload.py:488), but re-logging
|
||||
# the same inventory adds noise without information.
|
||||
shared.sd_model = sd_models.apply_balanced_offload(shared.sd_model, silent=True)
|
||||
devices.torch_gc(force=True, reason='ltx:base')
|
||||
t3 = time.time()
|
||||
timer.process.add('base', t2 - t1)
|
||||
timer.process.add('offload', t3 - t2)
|
||||
|
||||
if effective_upsample_enable and latents is not None:
|
||||
t4 = time.time()
|
||||
upsamplejob = shared.state.begin('Upsample')
|
||||
try:
|
||||
with video_utils.phase('Upsample'):
|
||||
t4 = time.time()
|
||||
# Shared-VAE exclude: both upsample pipes receive shared.sd_model.vae as a
|
||||
# constructor formality (pure latent -> latent forward). The main pipe already
|
||||
# owns the VAE's hook lifecycle, so walking it again here hits meta tensors
|
||||
# from the prior offload pass. Excluding also shortens the walk to the one
|
||||
# module that actually belongs to this pipe: latent_upsampler.
|
||||
upsample_exclude = ['vae']
|
||||
if latents.ndim == 4:
|
||||
latents = latents.unsqueeze(0)
|
||||
if caps.family == '0.9':
|
||||
global upsample_pipe # pylint: disable=global-statement
|
||||
upsample_pipe = load_upsample(upsample_pipe, upsample_repo_id_09)
|
||||
@@ -450,10 +445,7 @@ def run_ltx(task_id,
|
||||
'generator': get_generator(p.seed),
|
||||
'output_type': 'latent',
|
||||
}
|
||||
if latents.ndim == 4:
|
||||
latents = latents.unsqueeze(0)
|
||||
log.debug(f'Video: op=upsample family=0.9 latents={latents.shape} {up_args}')
|
||||
yield None, 'LTX: Upsample in progress...'
|
||||
latents = upsample_pipe(latents=latents, **up_args).frames[0]
|
||||
upsample_pipe = sd_models.apply_balanced_offload(upsample_pipe, exclude=upsample_exclude, silent=True)
|
||||
else:
|
||||
@@ -470,174 +462,141 @@ def run_ltx(task_id,
|
||||
'generator': get_generator(p.seed),
|
||||
'output_type': 'latent',
|
||||
}
|
||||
if latents.ndim == 4:
|
||||
latents = latents.unsqueeze(0)
|
||||
log.debug(f'Video: op=upsample family=2.x latents={latents.shape} auto={auto_refine_upsample} {up_args}')
|
||||
yield None, 'LTX: Upsample in progress...'
|
||||
latents = upsample_pipe_2x(latents=latents, **up_args).frames[0]
|
||||
upsample_pipe_2x = sd_models.apply_balanced_offload(upsample_pipe_2x, exclude=upsample_exclude, silent=True)
|
||||
except AssertionError as e:
|
||||
yield from abort(e, ok=True, p=p)
|
||||
return
|
||||
except Exception as e:
|
||||
yield from abort(e, ok=False, p=p)
|
||||
return
|
||||
t5 = time.time()
|
||||
timer.process.add('upsample', t5 - t4)
|
||||
shared.state.end(upsamplejob)
|
||||
t5 = time.time()
|
||||
timer.process.add('upsample', t5 - t4)
|
||||
|
||||
if refine_enable and latents is not None:
|
||||
t7 = time.time()
|
||||
refinejob = shared.state.begin('Refine')
|
||||
shared.sd_model = sd_models.apply_balanced_offload(shared.sd_model, silent=True)
|
||||
devices.torch_gc(force=True, reason='ltx:refine')
|
||||
# Refine is terminal: let the pipe decode internally so the final VAE pass runs inside
|
||||
# the same offload/cudnn context as a normal generation (matches Generic Video tab).
|
||||
refine_args = {
|
||||
'prompt_embeds': prompt_embeds,
|
||||
'prompt_attention_mask': prompt_attention_mask,
|
||||
'negative_prompt_embeds': negative_prompt_embeds,
|
||||
'negative_prompt_attention_mask': negative_prompt_attention_mask,
|
||||
'width': final_w,
|
||||
'height': final_h,
|
||||
'num_frames': get_frames(frames),
|
||||
'num_inference_steps': steps,
|
||||
'generator': get_generator(p.seed),
|
||||
'callback_on_step_end': diffusers_callback,
|
||||
'output_type': 'pil',
|
||||
}
|
||||
if p.cfg_scale is not None and p.cfg_scale > -1:
|
||||
refine_args['guidance_scale'] = p.cfg_scale
|
||||
if caps.supports_frame_rate_kwarg:
|
||||
refine_args['frame_rate'] = float(mp4_fps)
|
||||
if caps.supports_image_cond_noise_scale and image_cond_noise_scale is not None:
|
||||
refine_args['image_cond_noise_scale'] = image_cond_noise_scale
|
||||
if caps.supports_multi_condition and conditions_stage2:
|
||||
refine_args['conditions'] = conditions_stage2
|
||||
# Thread Stage-1 I2V init image through Stage 2 so first-frame identity survives refine.
|
||||
if caps.is_i2v and caps.repo_cls_name in ('LTXImageToVideoPipeline', 'LTX2ImageToVideoPipeline') and p.task_args.get('image') is not None:
|
||||
refine_args['image'] = p.task_args['image']
|
||||
if caps.family == '2.x':
|
||||
refine_args['use_cross_timestep'] = caps.use_cross_timestep
|
||||
# output_type='latent' skips the post-loop audio_vae + vocoder pass when audio
|
||||
# is unwanted; per-step audio cross-attention still runs for video conditioning.
|
||||
# Internal video decode is also skipped; vae_decode below picks it up.
|
||||
want_audio = caps.supports_audio and audio_enable
|
||||
if not want_audio:
|
||||
refine_args['output_type'] = 'latent'
|
||||
|
||||
saved_scheduler_stage2 = None
|
||||
try:
|
||||
with video_utils.phase('Refine'):
|
||||
t7 = time.time()
|
||||
shared.sd_model = sd_models.apply_balanced_offload(shared.sd_model, silent=True)
|
||||
devices.torch_gc(force=True, reason='ltx:refine')
|
||||
# Refine is terminal: let the pipe decode internally so the final VAE pass runs inside
|
||||
# the same offload/cudnn context as a normal generation (matches Generic Video tab).
|
||||
refine_args = {
|
||||
'prompt_embeds': prompt_embeds,
|
||||
'prompt_attention_mask': prompt_attention_mask,
|
||||
'negative_prompt_embeds': negative_prompt_embeds,
|
||||
'negative_prompt_attention_mask': negative_prompt_attention_mask,
|
||||
'width': final_w,
|
||||
'height': final_h,
|
||||
'num_frames': get_frames(frames),
|
||||
'num_inference_steps': steps,
|
||||
'generator': get_generator(p.seed),
|
||||
'callback_on_step_end': diffusers_callback,
|
||||
'output_type': 'pil',
|
||||
}
|
||||
if p.cfg_scale is not None and p.cfg_scale > -1:
|
||||
refine_args['guidance_scale'] = p.cfg_scale
|
||||
if caps.supports_frame_rate_kwarg:
|
||||
refine_args['frame_rate'] = float(mp4_fps)
|
||||
if caps.supports_image_cond_noise_scale and image_cond_noise_scale is not None:
|
||||
refine_args['image_cond_noise_scale'] = image_cond_noise_scale
|
||||
if caps.supports_multi_condition and conditions_stage2:
|
||||
refine_args['conditions'] = conditions_stage2
|
||||
# Thread Stage-1 I2V init image through Stage 2 so first-frame identity survives refine.
|
||||
if takes_init_image and p.task_args.get('image') is not None:
|
||||
refine_args['image'] = p.task_args['image']
|
||||
if caps.family == '2.x':
|
||||
# Stage 2 recipe (huggingface/diffusers#13217): fresh scheduler with shifting
|
||||
# disabled, 3 steps on STAGE_2_DISTILLED_SIGMA_VALUES, identity guidance.
|
||||
# Dev runs Distilled-on-Dev via the LoRA; Distilled is already at identity.
|
||||
from diffusers import FlowMatchEulerDiscreteScheduler
|
||||
saved_scheduler_stage2 = shared.sd_model.scheduler
|
||||
shared.sd_model.scheduler = FlowMatchEulerDiscreteScheduler.from_config(
|
||||
saved_scheduler_stage2.config,
|
||||
use_dynamic_shifting=False,
|
||||
shift_terminal=None,
|
||||
)
|
||||
if caps.supports_canonical_stage2:
|
||||
log.debug(f'LTX: stage=2 distilled=LoRA repo={caps.stage2_dev_lora_repo} weight={caps.stage2_dev_lora_weight}')
|
||||
offline_args = {'local_files_only': True} if shared.opts.offline_mode else {}
|
||||
# 2.5 keeps the LoRA in the model repo, so the file has to be named
|
||||
lora_args ={'weight_name': caps.stage2_dev_lora_weight} if caps.stage2_dev_lora_weight is not None else {}
|
||||
shared.sd_model.load_lora_weights(
|
||||
caps.stage2_dev_lora_repo,
|
||||
adapter_name=STAGE2_DEV_LORA_ADAPTER,
|
||||
cache_dir=shared.opts.hfcache_dir,
|
||||
**lora_args,
|
||||
**offline_args,
|
||||
)
|
||||
shared.sd_model.set_adapters([STAGE2_DEV_LORA_ADAPTER], [1.0])
|
||||
else:
|
||||
log.debug('LTX: stage=2 distilled=native')
|
||||
# Identity kwargs override any guidance left from earlier in refine_args.
|
||||
refine_args.update(_canonical_stage2_kwargs())
|
||||
refine_args.pop('num_inference_steps', None)
|
||||
elif caps.repo_cls_name == 'LTXConditionPipeline':
|
||||
refine_args['denoise_strength'] = refine_strength
|
||||
if latents.ndim == 4:
|
||||
latents = latents.unsqueeze(0)
|
||||
(
|
||||
refine_args['prompt_embeds'],
|
||||
refine_args['prompt_attention_mask'],
|
||||
refine_args['negative_prompt_embeds'],
|
||||
refine_args['negative_prompt_attention_mask'],
|
||||
) = _prompt_tensors_to_device(
|
||||
refine_args['prompt_embeds'],
|
||||
refine_args['prompt_attention_mask'],
|
||||
refine_args['negative_prompt_embeds'],
|
||||
refine_args['negative_prompt_attention_mask'],
|
||||
)
|
||||
log.debug(f'Video: op=refine cls={caps.repo_cls_name} latents={latents.shape} canonical_stage2={caps.supports_canonical_stage2}')
|
||||
yield None, 'LTX: Refine in progress...'
|
||||
refine_args['use_cross_timestep'] = caps.use_cross_timestep
|
||||
# output_type='latent' skips the post-loop audio_vae + vocoder pass when audio
|
||||
# is unwanted; per-step audio cross-attention still runs for video conditioning.
|
||||
# Internal video decode is also skipped; vae_decode below picks it up.
|
||||
want_audio = caps.supports_audio and audio
|
||||
if not want_audio:
|
||||
refine_args['output_type'] = 'latent'
|
||||
|
||||
saved_scheduler_stage2 = None
|
||||
try:
|
||||
if caps.family == '2.x':
|
||||
# Stage 2 recipe (huggingface/diffusers#13217): fresh scheduler with shifting
|
||||
# disabled, 3 steps on STAGE_2_DISTILLED_SIGMA_VALUES, identity guidance.
|
||||
# Dev runs Distilled-on-Dev via the LoRA; Distilled is already at identity.
|
||||
from diffusers import FlowMatchEulerDiscreteScheduler
|
||||
saved_scheduler_stage2 = shared.sd_model.scheduler
|
||||
shared.sd_model.scheduler = FlowMatchEulerDiscreteScheduler.from_config(
|
||||
saved_scheduler_stage2.config,
|
||||
use_dynamic_shifting=False,
|
||||
shift_terminal=None,
|
||||
)
|
||||
if caps.supports_canonical_stage2:
|
||||
log.debug(f'LTX: stage=2 distilled=LoRA repo={caps.stage2_dev_lora_repo} weight={caps.stage2_dev_lora_weight}')
|
||||
offline_args = {'local_files_only': True} if shared.opts.offline_mode else {}
|
||||
# 2.5 keeps the LoRA in the model repo, so the file has to be named
|
||||
lora_args ={'weight_name': caps.stage2_dev_lora_weight} if caps.stage2_dev_lora_weight is not None else {}
|
||||
shared.sd_model.load_lora_weights(
|
||||
caps.stage2_dev_lora_repo,
|
||||
adapter_name=STAGE2_DEV_LORA_ADAPTER,
|
||||
cache_dir=shared.opts.hfcache_dir,
|
||||
**lora_args,
|
||||
**offline_args,
|
||||
)
|
||||
shared.sd_model.set_adapters([STAGE2_DEV_LORA_ADAPTER], [1.0])
|
||||
else:
|
||||
log.debug('LTX: stage=2 distilled=native')
|
||||
# Identity kwargs override any guidance left from earlier in refine_args.
|
||||
refine_args.update(_canonical_stage2_kwargs())
|
||||
refine_args.pop('num_inference_steps', None)
|
||||
elif caps.repo_cls_name == 'LTXConditionPipeline':
|
||||
refine_args['denoise_strength'] = refine_strength
|
||||
if latents.ndim == 4:
|
||||
latents = latents.unsqueeze(0)
|
||||
(
|
||||
refine_args['prompt_embeds'],
|
||||
refine_args['prompt_attention_mask'],
|
||||
refine_args['negative_prompt_embeds'],
|
||||
refine_args['negative_prompt_attention_mask'],
|
||||
) = _prompt_tensors_to_device(
|
||||
refine_args['prompt_embeds'],
|
||||
refine_args['prompt_attention_mask'],
|
||||
refine_args['negative_prompt_embeds'],
|
||||
refine_args['negative_prompt_attention_mask'],
|
||||
)
|
||||
log.debug(f'Video: op=refine cls={caps.repo_cls_name} latents={latents.shape} canonical_stage2={caps.supports_canonical_stage2}')
|
||||
result = shared.sd_model(latents=latents, **refine_args)
|
||||
out = result.frames[0] if hasattr(result, 'frames') else None
|
||||
if want_audio:
|
||||
pixels = out
|
||||
if hasattr(result, 'audio') and result.audio is not None:
|
||||
audio = result.audio[0].float().cpu()
|
||||
audio_out = result.audio[0].float().cpu()
|
||||
latents = None
|
||||
else:
|
||||
latents = out
|
||||
except AssertionError as e:
|
||||
yield from abort(e, ok=True, p=p)
|
||||
return
|
||||
except Exception as e:
|
||||
yield from abort(e, ok=False, p=p)
|
||||
return
|
||||
finally:
|
||||
if saved_scheduler_stage2 is not None:
|
||||
if caps.supports_canonical_stage2:
|
||||
try:
|
||||
from modules.lora.extra_networks_lora import unload_diffusers
|
||||
unload_diffusers()
|
||||
except Exception as e:
|
||||
log.warning(f'LTX: stage=2 distilled=LoRA unload failed: {e}')
|
||||
shared.sd_model.scheduler = saved_scheduler_stage2
|
||||
# log.debug('LTX: stage=2 cleanup done')
|
||||
t8 = time.time()
|
||||
shared.sd_model = sd_models.apply_balanced_offload(shared.sd_model, silent=True)
|
||||
t9 = time.time()
|
||||
timer.process.add('refine', t8 - t7)
|
||||
timer.process.add('offload', t9 - t8)
|
||||
shared.state.end(refinejob)
|
||||
|
||||
if needs_latent_path:
|
||||
extra_networks.deactivate(p)
|
||||
finally:
|
||||
if saved_scheduler_stage2 is not None:
|
||||
if caps.supports_canonical_stage2:
|
||||
try:
|
||||
from modules.lora.extra_networks_lora import unload_diffusers
|
||||
unload_diffusers()
|
||||
except Exception as e:
|
||||
log.warning(f'LTX: stage=2 distilled=LoRA unload failed: {e}')
|
||||
shared.sd_model.scheduler = saved_scheduler_stage2
|
||||
t8 = time.time()
|
||||
shared.sd_model = sd_models.apply_balanced_offload(shared.sd_model, silent=True)
|
||||
t9 = time.time()
|
||||
timer.process.add('refine', t8 - t7)
|
||||
timer.process.add('offload', t9 - t8)
|
||||
|
||||
if needs_latent_path and latents is not None:
|
||||
# Decode any path that leaves latents intact: upsample-without-refine, or
|
||||
# refine with output_type='latent' (audio_enable=False).
|
||||
shared.sd_model = sd_models.apply_balanced_offload(shared.sd_model, exclude=['vae'], force=True, silent=True)
|
||||
devices.torch_gc(force=True, reason='ltx:vae')
|
||||
yield None, 'LTX: VAE decode in progress...'
|
||||
try:
|
||||
# refine with output_type='latent' (audio disabled).
|
||||
with video_utils.phase('VAE Decode'):
|
||||
shared.sd_model = sd_models.apply_balanced_offload(shared.sd_model, exclude=['vae'], force=True, silent=True)
|
||||
devices.torch_gc(force=True, reason='ltx:vae')
|
||||
if torch.is_tensor(latents):
|
||||
# 0.9.x returns raw latents with output_type='latent'; 2.x pre-denormalizes.
|
||||
frames_out = vae_decode(latents, decode_timestep if caps.supports_decode_timestep else 0.0, p.seed, denormalize=caps.family == '0.9')
|
||||
pixels = vae_decode(latents, decode_timestep if caps.supports_decode_timestep else 0.0, p.seed, denormalize=caps.family == '0.9')
|
||||
else:
|
||||
frames_out = latents
|
||||
except AssertionError as e:
|
||||
yield from abort(e, ok=True, p=p)
|
||||
return
|
||||
except Exception as e:
|
||||
yield from abort(e, ok=False, p=p)
|
||||
return
|
||||
pixels = frames_out
|
||||
t10 = time.time()
|
||||
shared.sd_model = sd_models.apply_balanced_offload(shared.sd_model, silent=True)
|
||||
t11 = time.time()
|
||||
timer.process.add('offload', t11 - t10)
|
||||
pixels = latents
|
||||
t10 = time.time()
|
||||
shared.sd_model = sd_models.apply_balanced_offload(shared.sd_model, silent=True)
|
||||
t11 = time.time()
|
||||
timer.process.add('offload', t11 - t10)
|
||||
|
||||
if not audio_enable:
|
||||
audio = None
|
||||
|
||||
aac_sample_rate = get_audio_rate(p)
|
||||
if not audio:
|
||||
audio_out = None
|
||||
|
||||
if mp4_interpolate > 0 and pixels is not None:
|
||||
p.video_interpolate = mp4_interpolate
|
||||
@@ -652,13 +611,14 @@ def run_ltx(task_id,
|
||||
x = apply_video_interpolation(p, x, count=mp4_interpolate)
|
||||
x = x * 2.0 - 1.0
|
||||
pixels = x.permute(1, 0, 2, 3).unsqueeze(0)
|
||||
|
||||
# LTX is conditioned on mp4_fps as the source rate; scale saved fps to keep duration constant
|
||||
from modules.processing_video import interpolation_factor
|
||||
save_fps = mp4_fps * interpolation_factor(p)
|
||||
num_frames, video_file, _thumb = save_video(
|
||||
num_frames, video_file, thumb_file = save_video(
|
||||
p=p,
|
||||
pixels=pixels,
|
||||
audio=audio,
|
||||
audio=audio_out,
|
||||
mp4_fps=save_fps,
|
||||
mp4_codec=mp4_codec,
|
||||
mp4_opt=mp4_opt,
|
||||
@@ -668,35 +628,154 @@ def run_ltx(task_id,
|
||||
mp4_frames=mp4_frames,
|
||||
mp4_thumb=mp4_thumb,
|
||||
mp4_interpolate=mp4_interpolate,
|
||||
aac_sample_rate=aac_sample_rate,
|
||||
aac_sample_rate=get_audio_rate(p),
|
||||
metadata={},
|
||||
)
|
||||
|
||||
t_end = time.time()
|
||||
if isinstance(pixels, list) and len(pixels) > 0 and isinstance(pixels[0], Image.Image):
|
||||
w, h = pixels[0].size
|
||||
elif hasattr(pixels, 'ndim') and pixels.ndim == 5:
|
||||
_n, _c, _t, h, w = pixels.shape
|
||||
elif hasattr(pixels, 'ndim') and pixels.ndim == 4:
|
||||
_n, h, w, _c = pixels.shape
|
||||
elif hasattr(pixels, 'shape'):
|
||||
h, w = pixels.shape[-2], pixels.shape[-1]
|
||||
else:
|
||||
w, h = p.width, p.height
|
||||
out_w, out_h = video_utils.pixel_size(pixels, fallback=(p.width, p.height))
|
||||
total_time = max(time.time() - t0, 1e-6)
|
||||
log.info(f'Processed: fn="{video_file}" frames={num_frames} fps={num_frames/total_time:.2f} its={p.steps/total_time:.3f} resolution={out_w}x{out_h} time={total_time:.2f} timers={timer.process.dct(no_total=True)} memory={memstats.memory_stats()}')
|
||||
# the decode paths never materialize PIL, so frames come back through the saved file
|
||||
images_out = pixels if isinstance(pixels, list) else []
|
||||
processed_out = processing.Processed(p, images_out, seed=p.seed, audio=audio_out)
|
||||
del pixels
|
||||
if audio is not None:
|
||||
del audio
|
||||
|
||||
resolution = f'{w}x{h}' if num_frames > 0 else None
|
||||
summary = timer.process.summary(min_time=0.25, total=False).replace('=', ' ')
|
||||
memory = shared.mem_mon.summary()
|
||||
total_time = max(t_end - t0, 1e-6)
|
||||
fps = f'{num_frames/total_time:.2f}'
|
||||
its = f'{(steps)/total_time:.3f}'
|
||||
|
||||
shared.state.end(videojob)
|
||||
progress.finish_task(task_id)
|
||||
return video_run.VideoResult(
|
||||
images=images_out,
|
||||
video_path=video_file,
|
||||
thumb_path=thumb_file,
|
||||
num_frames=num_frames,
|
||||
fps=float(save_fps),
|
||||
has_audio=audio_out is not None,
|
||||
still=False,
|
||||
processed=processed_out,
|
||||
width=out_w,
|
||||
height=out_h,
|
||||
)
|
||||
except video_run.VideoError:
|
||||
raise
|
||||
except AssertionError as e:
|
||||
# diffusers_callback raises this to unwind the denoise loop on interrupt
|
||||
log.info(f'Video: op=ltx {e}')
|
||||
raise video_run.VideoError('interrupted', 499) from e
|
||||
except Exception as e:
|
||||
log.error(f'Video: cls={shared.sd_model.__class__.__name__} op=ltx {e}')
|
||||
errors.display(e, 'LTX')
|
||||
raise video_run.VideoError(str(e), 500) from e
|
||||
finally:
|
||||
if p is not None:
|
||||
extra_networks.deactivate(p)
|
||||
p.close()
|
||||
|
||||
log.info(f'Processed: fn="{video_file}" frames={num_frames} fps={fps} its={its} resolution={resolution} time={t_end-t0:.2f} timers={timer.process.dct(no_total=True)} memory={memstats.memory_stats()}')
|
||||
yield video_file, f'Video | File {video_file} | Frames {num_frames} | Resolution {resolution} | f/s {fps} | it/s {its} ' + f"<div class='performance'><p>{summary} {memory}</p></div>"
|
||||
|
||||
def run_ltx(task_id,
|
||||
_ui_state,
|
||||
model: str,
|
||||
prompt: str,
|
||||
negative: str,
|
||||
styles: list,
|
||||
width: int,
|
||||
height: int,
|
||||
frames: int,
|
||||
auto_duration: bool,
|
||||
steps: int,
|
||||
sampler_index: int,
|
||||
guidance_scale: float,
|
||||
sampler_shift: float,
|
||||
dynamic_shift: bool,
|
||||
seed: int,
|
||||
upsample_enable: bool,
|
||||
upsample_ratio: float,
|
||||
refine_enable: bool,
|
||||
refine_strength: float,
|
||||
condition_strength: float,
|
||||
ltx_init_image,
|
||||
condition_last,
|
||||
condition_files,
|
||||
condition_video,
|
||||
condition_video_frames: int,
|
||||
condition_video_skip: int,
|
||||
decode_timestep: float,
|
||||
image_cond_noise_scale: float,
|
||||
mp4_fps: int,
|
||||
mp4_interpolate: int,
|
||||
mp4_codec: str,
|
||||
mp4_ext: str,
|
||||
mp4_opt: str,
|
||||
mp4_video: bool,
|
||||
mp4_frames: bool,
|
||||
mp4_sf: bool,
|
||||
mp4_thumb: bool,
|
||||
audio_enable: bool,
|
||||
_overrides,
|
||||
*args,
|
||||
**_kwargs,
|
||||
):
|
||||
# gradio adapter around run(): the signature is frozen since external callers bind to it by keyword
|
||||
progress.add_task_to_queue(task_id)
|
||||
with call_queue.get_lock():
|
||||
progress.start_task(task_id)
|
||||
memstats.reset_stats()
|
||||
timer.process.reset()
|
||||
yield None, 'LTX: Loading...'
|
||||
videojob = shared.state.begin('Video', task_id=task_id)
|
||||
shared.state.job_count = 1
|
||||
err = None
|
||||
res = None
|
||||
t0 = time.time()
|
||||
try:
|
||||
res = run(model,
|
||||
prompt=prompt,
|
||||
negative=negative,
|
||||
styles=styles,
|
||||
width=width,
|
||||
height=height,
|
||||
frames=frames,
|
||||
auto_duration=auto_duration,
|
||||
steps=steps,
|
||||
sampler_name=processing.get_sampler_name(sampler_index),
|
||||
sampler_shift=sampler_shift,
|
||||
dynamic_shift=dynamic_shift,
|
||||
seed=seed,
|
||||
guidance_scale=guidance_scale,
|
||||
upsample_enable=upsample_enable,
|
||||
upsample_ratio=upsample_ratio,
|
||||
refine_enable=refine_enable,
|
||||
refine_strength=refine_strength,
|
||||
condition_strength=condition_strength,
|
||||
init_image=ltx_init_image,
|
||||
condition_last=condition_last,
|
||||
condition_files=condition_files,
|
||||
condition_video=condition_video,
|
||||
condition_video_frames=condition_video_frames,
|
||||
condition_video_skip=condition_video_skip,
|
||||
decode_timestep=decode_timestep,
|
||||
image_cond_noise_scale=image_cond_noise_scale,
|
||||
audio=audio_enable,
|
||||
mp4_fps=mp4_fps,
|
||||
mp4_interpolate=mp4_interpolate,
|
||||
mp4_codec=mp4_codec,
|
||||
mp4_ext=mp4_ext,
|
||||
mp4_opt=mp4_opt,
|
||||
mp4_video=mp4_video,
|
||||
mp4_frames=mp4_frames,
|
||||
mp4_sf=mp4_sf,
|
||||
mp4_thumb=mp4_thumb,
|
||||
override_settings=_overrides,
|
||||
ui_state=_ui_state,
|
||||
script_args=args,
|
||||
)
|
||||
except video_run.VideoError as e:
|
||||
err = str(e)
|
||||
finally:
|
||||
shared.state.end(videojob)
|
||||
progress.finish_task(task_id)
|
||||
if res is None:
|
||||
yield None, f'LTX Error: {err}'
|
||||
return
|
||||
total_time = max(time.time() - t0, 1e-6)
|
||||
resolution = f'{res.width}x{res.height}' if res.num_frames > 0 else None
|
||||
fps = f'{res.num_frames/total_time:.2f}'
|
||||
its = f'{res.processed.steps/total_time:.3f}'
|
||||
summary = timer.process.summary(min_time=0.25, total=False).replace('=', ' ')
|
||||
memory = shared.mem_mon.summary()
|
||||
yield res.video_path, f'Video | File {res.video_path} | Frames {res.num_frames} | Resolution {resolution} | f/s {fps} | it/s {its} ' + f"<div class='performance'><p>{summary} {memory}</p></div>"
|
||||
|
||||
+20
-13
@@ -65,6 +65,9 @@ def load_upsample(upsample_pipe, upsample_repo_id):
|
||||
cache_dir=shared.opts.hfcache_dir,
|
||||
torch_dtype=devices.dtype,
|
||||
)
|
||||
# only the upsampler, since the pipe borrows the model's vae and moving the whole pipe
|
||||
# would drag that along into the meta tensors the caller's offload exclude avoids
|
||||
upsample_pipe.latent_upsampler.to(devices.device)
|
||||
upsample_pipe.sdnext_upsample_repo = upsample_repo_id
|
||||
t1 = time.time()
|
||||
timer.process.add('load', t1 - t0)
|
||||
@@ -180,6 +183,20 @@ def make_condition(condition_cls, family: str, frames, strength: float, is_video
|
||||
return condition_cls(image=frames, frame_index=index, strength=strength)
|
||||
|
||||
|
||||
def open_condition(src) -> Image.Image:
|
||||
"""A conditioning source as a PIL image, from a gradio upload handle or from an api reference.
|
||||
|
||||
A string goes to the api decoder, which reads base64 and upload refs, rather than being opened
|
||||
as a path: naming a file is the caller's way of reading one it never uploaded.
|
||||
"""
|
||||
if hasattr(src, 'name'):
|
||||
return Image.open(src.name)
|
||||
if isinstance(src, str):
|
||||
from modules.api.api import decode_base64_to_image
|
||||
return decode_base64_to_image(src)
|
||||
return src
|
||||
|
||||
|
||||
def get_conditions(width, height, condition_strength, condition_images, condition_files, condition_video, condition_video_frames, condition_video_skip, family: str = '0.9', num_frames=None, condition_last=None):
|
||||
condition_cls = _condition_cls(family)
|
||||
if condition_cls is None:
|
||||
@@ -188,10 +205,7 @@ def get_conditions(width, height, condition_strength, condition_images, conditio
|
||||
if condition_images is not None:
|
||||
for condition_image in condition_images:
|
||||
try:
|
||||
if isinstance(condition_image, str):
|
||||
from modules.api.api import decode_base64_to_image
|
||||
condition_image = decode_base64_to_image(condition_image)
|
||||
condition_image = condition_image.convert('RGB').resize((width, height), resample=Image.Resampling.LANCZOS)
|
||||
condition_image = open_condition(condition_image).convert('RGB').resize((width, height), resample=Image.Resampling.LANCZOS)
|
||||
conditions.append(make_condition(condition_cls, family, condition_image, condition_strength, is_video=False))
|
||||
log.debug(f'Video condition: family={family} image={condition_image.size} strength={condition_strength}')
|
||||
except Exception as e:
|
||||
@@ -200,11 +214,7 @@ def get_conditions(width, height, condition_strength, condition_images, conditio
|
||||
batch_images = []
|
||||
for fn in condition_files:
|
||||
try:
|
||||
if hasattr(fn, 'name'):
|
||||
condition_image = Image.open(fn.name).convert('RGB').resize((width, height), resample=Image.Resampling.LANCZOS)
|
||||
else:
|
||||
condition_image = fn.convert('RGB').resize((width, height), resample=Image.Resampling.LANCZOS)
|
||||
batch_images.append(condition_image)
|
||||
batch_images.append(open_condition(fn).convert('RGB').resize((width, height), resample=Image.Resampling.LANCZOS))
|
||||
except Exception as e:
|
||||
log.error(f'LTX condition files: {e}')
|
||||
if len(batch_images) > 0:
|
||||
@@ -222,10 +232,7 @@ def get_conditions(width, height, condition_strength, condition_images, conditio
|
||||
log.error(f'LTX condition video: {e}')
|
||||
if condition_last is not None:
|
||||
try:
|
||||
if isinstance(condition_last, str):
|
||||
from modules.api.api import decode_base64_to_image
|
||||
condition_last = decode_base64_to_image(condition_last)
|
||||
condition_last = condition_last.convert('RGB').resize((width, height), resample=Image.Resampling.LANCZOS)
|
||||
condition_last = open_condition(condition_last).convert('RGB').resize((width, height), resample=Image.Resampling.LANCZOS)
|
||||
# 2.x reads index as a latent index and accepts -1 for the final frame; 0.9 uses a pixel index.
|
||||
last_index = -1 if family == '2.x' else max((num_frames or 1) - 1, 0)
|
||||
conditions.append(make_condition(condition_cls, family, condition_last, condition_strength, is_video=False, index=last_index))
|
||||
|
||||
@@ -29,6 +29,16 @@ class VideoResult:
|
||||
has_audio: bool
|
||||
still: bool
|
||||
processed: processing.Processed
|
||||
width: int = 0 # what was generated, which is not what was requested whenever a runner rounds or a model picks
|
||||
height: int = 0
|
||||
|
||||
|
||||
def normalize_override_settings(override_settings):
|
||||
"""Override settings as a dict. The ui control emits "setting: value" pairs; api callers send the dict."""
|
||||
if isinstance(override_settings, (list, tuple)):
|
||||
from modules.generation_parameters_copypaste import create_override_settings_dict
|
||||
return create_override_settings_dict(override_settings)
|
||||
return override_settings
|
||||
|
||||
|
||||
def resolve_model(engine: str | None, model: str | None) -> tuple[models_def.Model, bool]:
|
||||
@@ -138,9 +148,7 @@ def run(selected: models_def.Model, *,
|
||||
debug('Video: model still not loaded')
|
||||
raise VideoError('model not loaded', 500)
|
||||
|
||||
if isinstance(override_settings, (list, tuple)): # the ui override control emits "setting: value" pairs; always empty on the video tab since the control stays hidden
|
||||
from modules.generation_parameters_copypaste import create_override_settings_dict
|
||||
override_settings = create_override_settings_dict(override_settings)
|
||||
override_settings = normalize_override_settings(override_settings) # always empty on the video tab, since the control stays hidden
|
||||
|
||||
p = processing.StableDiffusionProcessingVideo(
|
||||
sd_model=shared.sd_model,
|
||||
@@ -302,7 +310,8 @@ def run(selected: models_def.Model, *,
|
||||
|
||||
if getattr(p, 'video_still', False):
|
||||
stills = processed.images[:1] # already trimmed in process_decode; defensive
|
||||
return VideoResult(images=stills, video_path=None, thumb_path=None, num_frames=len(stills), fps=0.0, has_audio=False, still=True, processed=processed)
|
||||
still_w, still_h = video_utils.pixel_size(stills, fallback=(p.width, p.height))
|
||||
return VideoResult(images=stills, video_path=None, thumb_path=None, num_frames=len(stills), fps=0.0, has_audio=False, still=True, processed=processed, width=still_w, height=still_h)
|
||||
|
||||
if hasattr(processed, 'images') and processed.images is not None:
|
||||
pixels = video_save.images_to_tensor(processed.images)
|
||||
@@ -341,8 +350,9 @@ def run(selected: models_def.Model, *,
|
||||
mp4_interpolate=mp4_interpolate,
|
||||
metadata={},
|
||||
)
|
||||
out_w, out_h = video_utils.pixel_size(processed.images, fallback=(p.width, p.height))
|
||||
del pixels
|
||||
return VideoResult(images=processed.images, video_path=video_file, thumb_path=thumb_file, num_frames=num_frames, fps=float(save_fps), has_audio=waveform is not None, still=False, processed=processed)
|
||||
return VideoResult(images=processed.images, video_path=video_file, thumb_path=thumb_file, num_frames=num_frames, fps=float(save_fps), has_audio=waveform is not None, still=False, processed=processed, width=out_w, height=out_h)
|
||||
|
||||
|
||||
def generate(task_id, ui_state,
|
||||
|
||||
@@ -3,6 +3,7 @@ import sys
|
||||
import time
|
||||
import inspect
|
||||
import importlib.util
|
||||
from contextlib import contextmanager
|
||||
from dataclasses import dataclass
|
||||
from PIL import Image
|
||||
from installer import install
|
||||
@@ -73,6 +74,36 @@ def has_torchaudio():
|
||||
return False
|
||||
|
||||
|
||||
@contextmanager
|
||||
def phase(title: str):
|
||||
"""Scoped generation phase, so an abort cannot leave the job holding a stage that never ended."""
|
||||
jobid = shared.state.begin(title)
|
||||
try:
|
||||
yield jobid
|
||||
finally:
|
||||
shared.state.end(jobid)
|
||||
|
||||
|
||||
def pixel_size(pixels, fallback: tuple[int, int] = (0, 0)) -> tuple[int, int]:
|
||||
"""Width and height of decoded frames, whether they arrive as PIL images or as a tensor.
|
||||
|
||||
The runners request a resolution and the model answers with another one often enough that
|
||||
the request is not a usable substitute, so read it off the pixels and keep the fallback for
|
||||
the case where nothing was decoded at all.
|
||||
"""
|
||||
if isinstance(pixels, list):
|
||||
return pixels[0].size if len(pixels) > 0 and hasattr(pixels[0], 'size') else fallback
|
||||
ndim = getattr(pixels, 'ndim', None)
|
||||
if ndim == 5: # NCTHW
|
||||
return int(pixels.shape[-1]), int(pixels.shape[-2])
|
||||
if ndim == 4: # NHWC
|
||||
return int(pixels.shape[2]), int(pixels.shape[1])
|
||||
shape = getattr(pixels, 'shape', None)
|
||||
if shape is not None and len(shape) >= 2:
|
||||
return int(shape[-1]), int(shape[-2])
|
||||
return fallback
|
||||
|
||||
|
||||
def classify_extension(fn: str):
|
||||
"""Media kind of a filename, None when the extension is not one sdnext reads."""
|
||||
lower = str(fn).lower()
|
||||
|
||||
@@ -0,0 +1,443 @@
|
||||
#!/usr/bin/env python
|
||||
"""
|
||||
Offline unit tests for the LTX keyword core in modules.ltx.ltx_process.
|
||||
|
||||
``run_ltx`` used to be the only way to generate with LTX: a gradio generator that reported
|
||||
failure by yielding an error string, which is why LTX had no API. ``run`` is the keyword core
|
||||
underneath it, converged on ``video_run.run``'s names, result and error protocol.
|
||||
|
||||
Covers:
|
||||
|
||||
- ``run`` is keyword-only past the model name, so a caller cannot mis-bind by position
|
||||
- ``run_ltx``'s positional signature, parameter for parameter, since external callers bind to
|
||||
it by keyword and a rename would silently drop an argument into **kwargs
|
||||
- the rejections that happen before anything is loaded, and that they do not load
|
||||
- the adapter's delegation: the shapes it yields on success and on a typed failure
|
||||
- ``open_condition`` resolving a string through the api decoder rather than as a path
|
||||
- ``pixel_size`` over the frame shapes the two paths produce
|
||||
- ``phase`` ending its job when the body raises
|
||||
|
||||
No running server required.
|
||||
|
||||
Usage:
|
||||
python test/test-ltx-core.py
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import inspect
|
||||
|
||||
script_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
sys.path.insert(0, script_dir)
|
||||
os.chdir(script_dir)
|
||||
|
||||
os.environ['SD_INSTALL_QUIET'] = '1'
|
||||
|
||||
# Bootstrap cmd_args before any module that pulls in shared.py.
|
||||
import modules.cmd_args # pylint: disable=wrong-import-position
|
||||
import installer # pylint: disable=wrong-import-position
|
||||
orig_argv = sys.argv
|
||||
sys.argv = [sys.argv[0]]
|
||||
try:
|
||||
modules.cmd_args.parse_args()
|
||||
finally:
|
||||
sys.argv = orig_argv
|
||||
installer.add_args(modules.cmd_args.parser)
|
||||
modules.cmd_args.parsed, _ = modules.cmd_args.parser.parse_known_args([])
|
||||
|
||||
from modules.errors import log # pylint: disable=wrong-import-position
|
||||
from modules.video_models import video_run, video_utils # pylint: disable=wrong-import-position
|
||||
|
||||
|
||||
results: dict[str, dict] = {}
|
||||
|
||||
# Frozen because callers outside this repo bind every one of these by keyword.
|
||||
ADAPTER_PARAMS = [
|
||||
'task_id', '_ui_state', 'model', 'prompt', 'negative', 'styles', 'width', 'height', 'frames',
|
||||
'auto_duration', 'steps', 'sampler_index', 'guidance_scale', 'sampler_shift', 'dynamic_shift',
|
||||
'seed', 'upsample_enable', 'upsample_ratio', 'refine_enable', 'refine_strength',
|
||||
'condition_strength', 'ltx_init_image', 'condition_last', 'condition_files', 'condition_video',
|
||||
'condition_video_frames', 'condition_video_skip', 'decode_timestep', 'image_cond_noise_scale',
|
||||
'mp4_fps', 'mp4_interpolate', 'mp4_codec', 'mp4_ext', 'mp4_opt', 'mp4_video', 'mp4_frames',
|
||||
'mp4_sf', 'mp4_thumb', 'audio_enable', '_overrides',
|
||||
]
|
||||
|
||||
# The names the core answers to. Divergence from video_run.run is deliberate only where LTX has
|
||||
# no equivalent concept, so a new name appearing here should be a considered choice.
|
||||
CORE_PARAMS = [
|
||||
'prompt', 'negative', 'styles', 'width', 'height', 'frames', 'auto_duration', 'steps',
|
||||
'sampler_name', 'sampler_shift', 'dynamic_shift', 'seed', 'guidance_scale',
|
||||
'upsample_enable', 'upsample_ratio', 'refine_enable', 'refine_strength', 'condition_strength',
|
||||
'init_image', 'condition_last', 'condition_files', 'condition_video', 'condition_video_frames',
|
||||
'condition_video_skip', 'decode_timestep', 'image_cond_noise_scale', 'audio',
|
||||
'mp4_fps', 'mp4_interpolate', 'mp4_codec', 'mp4_ext', 'mp4_opt', 'mp4_video', 'mp4_frames',
|
||||
'mp4_sf', 'mp4_thumb', 'override_settings', 'ui_state', 'scripts', 'script_args',
|
||||
'per_script_args', 'extra_p',
|
||||
]
|
||||
|
||||
|
||||
def category(name: str):
|
||||
if name not in results:
|
||||
results[name] = {'passed': 0, 'failed': 0, 'skipped': 0, 'tests': []}
|
||||
return name
|
||||
|
||||
|
||||
def record(cat: str, passed: bool, name: str, detail: str = ''):
|
||||
status = 'PASS' if passed else 'FAIL'
|
||||
results[cat]['passed' if passed else 'failed'] += 1
|
||||
results[cat]['tests'].append((status, name))
|
||||
msg = f' {status}: {name}'
|
||||
if detail:
|
||||
msg += f' ({detail})'
|
||||
if passed:
|
||||
log.info(msg)
|
||||
else:
|
||||
log.error(msg)
|
||||
|
||||
|
||||
def skip(cat: str, name: str, reason: str):
|
||||
results[cat]['skipped'] += 1
|
||||
results[cat]['tests'].append(('SKIP', name))
|
||||
log.warning(f' SKIP: {name} ({reason})')
|
||||
|
||||
|
||||
def run_test(cat: str, fn):
|
||||
name = fn.__name__
|
||||
try:
|
||||
ok = fn()
|
||||
if ok is False:
|
||||
record(cat, False, name)
|
||||
elif isinstance(ok, str):
|
||||
skip(cat, name, ok)
|
||||
else:
|
||||
record(cat, True, name)
|
||||
except AssertionError as e:
|
||||
record(cat, False, name, str(e))
|
||||
except Exception as e: # pylint: disable=broad-except
|
||||
record(cat, False, name, f'exception: {type(e).__name__}: {e}')
|
||||
|
||||
|
||||
def ltx():
|
||||
from modules.ltx import ltx_process
|
||||
return ltx_process
|
||||
|
||||
|
||||
# --- signatures -------------------------------------------------------------------------------
|
||||
|
||||
def test_core_takes_only_model_positionally():
|
||||
sig = inspect.signature(ltx().run)
|
||||
positional = [n for n, prm in sig.parameters.items() if prm.kind is prm.POSITIONAL_OR_KEYWORD]
|
||||
assert positional == ['model'], f'positional params: {positional}'
|
||||
return True
|
||||
|
||||
|
||||
def test_core_parameter_names():
|
||||
sig = inspect.signature(ltx().run)
|
||||
kwonly = [n for n, prm in sig.parameters.items() if prm.kind is prm.KEYWORD_ONLY]
|
||||
assert kwonly == CORE_PARAMS, f'unexpected core signature: {kwonly}'
|
||||
return True
|
||||
|
||||
|
||||
def test_core_returns_video_result():
|
||||
sig = inspect.signature(ltx().run)
|
||||
assert sig.return_annotation is video_run.VideoResult, f'return annotation: {sig.return_annotation}'
|
||||
return True
|
||||
|
||||
|
||||
def test_adapter_signature_is_frozen():
|
||||
sig = inspect.signature(ltx().run_ltx)
|
||||
named = [n for n, prm in sig.parameters.items() if prm.kind is prm.POSITIONAL_OR_KEYWORD]
|
||||
assert named == ADAPTER_PARAMS, f'adapter signature drifted: {named}'
|
||||
return True
|
||||
|
||||
|
||||
def test_adapter_still_absorbs_extra_arguments():
|
||||
sig = inspect.signature(ltx().run_ltx)
|
||||
kinds = {prm.kind for prm in sig.parameters.values()}
|
||||
assert inspect.Parameter.VAR_POSITIONAL in kinds, 'adapter must keep *args for the script vector'
|
||||
assert inspect.Parameter.VAR_KEYWORD in kinds, 'adapter must keep **kwargs so callers survive signature growth'
|
||||
return True
|
||||
|
||||
|
||||
def test_adapter_is_a_generator():
|
||||
assert inspect.isgeneratorfunction(ltx().run_ltx), 'the tab binds to a generator'
|
||||
assert not inspect.isgeneratorfunction(ltx().run), 'the core returns a value rather than yielding'
|
||||
return True
|
||||
|
||||
|
||||
# --- rejections -------------------------------------------------------------------------------
|
||||
|
||||
def rejects(**kwargs) -> video_run.VideoError:
|
||||
"""Call the core with the loader poisoned, and return the error it raised."""
|
||||
ltx_process = ltx()
|
||||
orig_load = ltx_process.load_model
|
||||
ltx_process.load_model = lambda *a, **k: (_ for _ in ()).throw(AssertionError('loader must not run'))
|
||||
try:
|
||||
ltx_process.run(**kwargs)
|
||||
except video_run.VideoError as e:
|
||||
return e
|
||||
finally:
|
||||
ltx_process.load_model = orig_load
|
||||
raise AssertionError(f'no VideoError raised for {kwargs}')
|
||||
|
||||
|
||||
def test_empty_model_rejected_before_load():
|
||||
for model in (None, '', 'None'):
|
||||
err = rejects(model=model, prompt='test')
|
||||
assert err.code == 400, f'model={model!r} code={err.code}'
|
||||
return True
|
||||
|
||||
|
||||
def test_separator_rejected_before_load():
|
||||
err = rejects(model='─────── LTX-2.5 ───────', prompt='test')
|
||||
assert err.code == 400, f'code={err.code}'
|
||||
assert 'separator' in str(err), f'message={err}'
|
||||
return True
|
||||
|
||||
|
||||
def test_missing_av_rejected_before_load():
|
||||
ltx_process = ltx()
|
||||
orig_check = video_utils.check_av
|
||||
video_utils.check_av = lambda: None
|
||||
try:
|
||||
err = rejects(model='LTXVideo 0.9.6 2B T2V', prompt='test', mp4_video=True)
|
||||
assert err.code == 500, f'code={err.code}'
|
||||
assert 'av' in str(err), f'message={err}'
|
||||
finally:
|
||||
video_utils.check_av = orig_check
|
||||
assert ltx_process.run is not None
|
||||
return True
|
||||
|
||||
|
||||
def test_missing_av_ignored_when_no_video_wanted():
|
||||
"""Frames-only output does not need an encoder, so the check must not reject it."""
|
||||
ltx_process = ltx()
|
||||
orig_check = video_utils.check_av
|
||||
calls = []
|
||||
video_utils.check_av = lambda: calls.append(1)
|
||||
orig_load = ltx_process.load_model
|
||||
ltx_process.load_model = lambda *a, **k: (_ for _ in ()).throw(RuntimeError('reached the loader'))
|
||||
try:
|
||||
ltx_process.run(model='LTXVideo 0.9.6 2B T2V', prompt='test', mp4_video=False, mp4_frames=True)
|
||||
except video_run.VideoError as e:
|
||||
raise AssertionError(f'rejected before the loader: {e}') from e
|
||||
except RuntimeError:
|
||||
pass # got past the checks, which is the point
|
||||
finally:
|
||||
video_utils.check_av = orig_check
|
||||
ltx_process.load_model = orig_load
|
||||
assert len(calls) == 0, 'av was probed for a run that saves no video'
|
||||
return True
|
||||
|
||||
|
||||
# --- adapter delegation -----------------------------------------------------------------------
|
||||
|
||||
def fake_result(**kwargs):
|
||||
class FakeProcessed: # pylint: disable=too-few-public-methods
|
||||
steps = 8
|
||||
defaults = dict(
|
||||
images=[], video_path='/tmp/fake.mp4', thumb_path=None, num_frames=17, fps=24.0,
|
||||
has_audio=False, still=False, processed=FakeProcessed(), width=768, height=512,
|
||||
)
|
||||
defaults.update(kwargs)
|
||||
return video_run.VideoResult(**defaults)
|
||||
|
||||
|
||||
def drive_adapter(core):
|
||||
"""Run the adapter end to end with the core replaced, and return what it yielded."""
|
||||
ltx_process = ltx()
|
||||
orig_run = ltx_process.run
|
||||
ltx_process.run = core
|
||||
try:
|
||||
gen = ltx_process.run_ltx(
|
||||
task_id='', _ui_state='', model='LTXVideo 0.9.6 2B T2V', prompt='test', negative='',
|
||||
styles=[], width=768, height=512, frames=17, auto_duration=False, steps=8,
|
||||
sampler_index=0, guidance_scale=1.0, sampler_shift=-1.0, dynamic_shift=False, seed=-1,
|
||||
upsample_enable=False, upsample_ratio=2.0, refine_enable=False, refine_strength=0.4,
|
||||
condition_strength=1.0, ltx_init_image=None, condition_last=None, condition_files=None,
|
||||
condition_video=None, condition_video_frames=-1, condition_video_skip=0,
|
||||
decode_timestep=0.05, image_cond_noise_scale=0.025, mp4_fps=24, mp4_interpolate=0,
|
||||
mp4_codec='libx264', mp4_ext='mp4', mp4_opt='crf=16', mp4_video=True, mp4_frames=False,
|
||||
mp4_sf=False, mp4_thumb=True, audio_enable=False, _overrides={},
|
||||
)
|
||||
return list(gen)
|
||||
finally:
|
||||
ltx_process.run = orig_run
|
||||
|
||||
|
||||
def test_adapter_yields_the_video_path_on_success():
|
||||
seen = {}
|
||||
|
||||
def core(model, **kwargs):
|
||||
seen['model'] = model
|
||||
seen['kwargs'] = kwargs
|
||||
return fake_result()
|
||||
|
||||
try:
|
||||
yields = drive_adapter(core)
|
||||
except Exception as e: # pylint: disable=broad-except
|
||||
return f'adapter needs more runtime state than this harness provides: {type(e).__name__}: {e}'
|
||||
assert len(yields) == 2, f'expected a loading yield and a final yield, got {len(yields)}'
|
||||
assert yields[0][0] is None, f'first yield should carry no file: {yields[0]}'
|
||||
assert yields[-1][0] == '/tmp/fake.mp4', f'final yield: {yields[-1]}'
|
||||
assert '768x512' in yields[-1][1], f'resolution missing from the summary: {yields[-1][1]}'
|
||||
return True
|
||||
|
||||
|
||||
def test_adapter_renames_into_the_core():
|
||||
seen = {}
|
||||
|
||||
def core(model, **kwargs):
|
||||
seen['model'] = model
|
||||
seen['kwargs'] = kwargs
|
||||
return fake_result()
|
||||
|
||||
try:
|
||||
drive_adapter(core)
|
||||
except Exception as e: # pylint: disable=broad-except
|
||||
return f'adapter needs more runtime state than this harness provides: {type(e).__name__}: {e}'
|
||||
kwargs = seen.get('kwargs', {})
|
||||
assert seen.get('model') == 'LTXVideo 0.9.6 2B T2V', f'model not passed positionally: {seen}'
|
||||
assert 'init_image' in kwargs and 'ltx_init_image' not in kwargs, 'init image not renamed'
|
||||
assert 'audio' in kwargs and 'audio_enable' not in kwargs, 'audio flag not renamed'
|
||||
assert 'sampler_name' in kwargs and 'sampler_index' not in kwargs, 'sampler not resolved to a name'
|
||||
assert 'override_settings' in kwargs, 'overrides not forwarded'
|
||||
assert 'ui_state' in kwargs, 'ui state not forwarded'
|
||||
return True
|
||||
|
||||
|
||||
def test_adapter_reports_a_typed_failure_as_text():
|
||||
def core(model, **kwargs): # pylint: disable=unused-argument
|
||||
raise video_run.VideoError('no model selected', 400)
|
||||
|
||||
try:
|
||||
yields = drive_adapter(core)
|
||||
except Exception as e: # pylint: disable=broad-except
|
||||
return f'adapter needs more runtime state than this harness provides: {type(e).__name__}: {e}'
|
||||
assert yields[-1][0] is None, f'a failure must not report a file: {yields[-1]}'
|
||||
assert yields[-1][1].startswith('LTX Error:'), f'final yield: {yields[-1]}'
|
||||
assert 'no model selected' in yields[-1][1], f'message lost: {yields[-1][1]}'
|
||||
return True
|
||||
|
||||
|
||||
# --- helpers ----------------------------------------------------------------------------------
|
||||
|
||||
def test_open_condition_does_not_read_paths():
|
||||
"""A string names an upload or carries base64; opening it as a path would read any file."""
|
||||
from modules.ltx import ltx_util
|
||||
target = os.path.join(script_dir, 'requirements.txt')
|
||||
if not os.path.exists(target):
|
||||
return 'no readable file in the repo root to probe with'
|
||||
try:
|
||||
ltx_util.open_condition(target)
|
||||
except Exception: # pylint: disable=broad-except
|
||||
return True # the decoder rejected it, which is the point
|
||||
raise AssertionError('a filesystem path was accepted as a conditioning source')
|
||||
|
||||
|
||||
def test_open_condition_passes_images_through():
|
||||
from PIL import Image
|
||||
from modules.ltx import ltx_util
|
||||
img = Image.new('RGB', (8, 8))
|
||||
assert ltx_util.open_condition(img) is img
|
||||
return True
|
||||
|
||||
|
||||
def test_pixel_size_over_both_frame_shapes():
|
||||
import torch
|
||||
from PIL import Image
|
||||
assert video_utils.pixel_size([Image.new('RGB', (640, 352))]) == (640, 352)
|
||||
assert video_utils.pixel_size(torch.zeros(1, 3, 17, 352, 640)) == (640, 352)
|
||||
assert video_utils.pixel_size(torch.zeros(17, 352, 640, 3)) == (640, 352)
|
||||
return True
|
||||
|
||||
|
||||
def test_pixel_size_falls_back_when_nothing_decoded():
|
||||
assert video_utils.pixel_size([], fallback=(768, 512)) == (768, 512)
|
||||
assert video_utils.pixel_size(None, fallback=(768, 512)) == (768, 512)
|
||||
assert video_utils.pixel_size(None) == (0, 0)
|
||||
return True
|
||||
|
||||
|
||||
def test_phase_ends_its_job_when_the_body_raises():
|
||||
from modules import shared
|
||||
before = shared.state.job
|
||||
try:
|
||||
with video_utils.phase('TestPhase'):
|
||||
raise RuntimeError('boom')
|
||||
except RuntimeError:
|
||||
pass
|
||||
assert shared.state.job == before, f'phase leaked: job={shared.state.job!r} expected={before!r}'
|
||||
return True
|
||||
|
||||
|
||||
def test_video_result_resolution_defaults_to_zero():
|
||||
res = video_run.VideoResult(images=[], video_path=None, thumb_path=None, num_frames=0,
|
||||
fps=0.0, has_audio=False, still=False, processed=None)
|
||||
assert (res.width, res.height) == (0, 0), f'{res.width}x{res.height}'
|
||||
return True
|
||||
|
||||
|
||||
def run_all():
|
||||
log.warning('=== LTX keyword core ===')
|
||||
|
||||
cat = category('signatures')
|
||||
for fn in [
|
||||
test_core_takes_only_model_positionally,
|
||||
test_core_parameter_names,
|
||||
test_core_returns_video_result,
|
||||
test_adapter_signature_is_frozen,
|
||||
test_adapter_still_absorbs_extra_arguments,
|
||||
test_adapter_is_a_generator,
|
||||
]:
|
||||
run_test(cat, fn)
|
||||
|
||||
cat = category('rejections')
|
||||
for fn in [
|
||||
test_empty_model_rejected_before_load,
|
||||
test_separator_rejected_before_load,
|
||||
test_missing_av_rejected_before_load,
|
||||
test_missing_av_ignored_when_no_video_wanted,
|
||||
]:
|
||||
run_test(cat, fn)
|
||||
|
||||
cat = category('adapter')
|
||||
for fn in [
|
||||
test_adapter_yields_the_video_path_on_success,
|
||||
test_adapter_renames_into_the_core,
|
||||
test_adapter_reports_a_typed_failure_as_text,
|
||||
]:
|
||||
run_test(cat, fn)
|
||||
|
||||
cat = category('helpers')
|
||||
for fn in [
|
||||
test_open_condition_does_not_read_paths,
|
||||
test_open_condition_passes_images_through,
|
||||
test_pixel_size_over_both_frame_shapes,
|
||||
test_pixel_size_falls_back_when_nothing_decoded,
|
||||
test_phase_ends_its_job_when_the_body_raises,
|
||||
test_video_result_resolution_defaults_to_zero,
|
||||
]:
|
||||
run_test(cat, fn)
|
||||
|
||||
log.warning('=== Results ===')
|
||||
total_passed = 0
|
||||
total_failed = 0
|
||||
total_skipped = 0
|
||||
for cat_name, info in results.items():
|
||||
ok = info['failed'] == 0
|
||||
status = 'PASS' if ok else 'FAIL'
|
||||
log.info(f" {cat_name}: {info['passed']} passed, {info['failed']} failed, {info['skipped']} skipped [{status}]")
|
||||
total_passed += info['passed']
|
||||
total_failed += info['failed']
|
||||
total_skipped += info['skipped']
|
||||
log.warning(f'Total: {total_passed} passed, {total_failed} failed, {total_skipped} skipped')
|
||||
return total_failed == 0
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
import time
|
||||
t0 = time.time()
|
||||
success = run_all()
|
||||
log.warning(f'Total time: {time.time() - t0:.2f}s')
|
||||
sys.exit(0 if success else 1)
|
||||
Reference in New Issue
Block a user