Public/automatic
1
0
Fork 0
mirror of https://github.com/vladmandic/automatic synced 2026-09-19 17:24:32 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat(ltx): add a keyword core to the ltx runner

Browse Source 
run_ltx reported failure by yielding a string, which is why LTX had no API.
run() is the core underneath: keyword arguments named as video_run.run names
them, a VideoResult back, VideoError out with 499 for an interrupt. The lock,
progress and summary stay in the adapter, whose signature is unchanged since
callers bind to it by keyword. Failure now closes the processing object and
deactivates networks, which abort never did.
This commit is contained in:
CalamitousFelicitousness
2026-08-19 01:39:45 +01:00
parent bd0ea5ee1b
commit 8d69b011e1
4 changed files with 841 additions and 312 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/ltx/ltx_process.py
+378 -309
View File
@@ -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,82 @@ 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 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
raise video_run.VideoError('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)
raise video_run.VideoError('dropdown separator selected, pick an actual model below', 400)
if mp4_video and video_utils.check_av() is None:
raise video_run.VideoError('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'):
raise video_run.VideoError(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:
raise video_run.VideoError('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 +229,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 +255,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 +265,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 +309,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 +332,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 +391,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:
raise video_run.VideoError('interrupted', 499)
raise video_run.VideoError('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 +435,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 +452,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 +601,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 +618,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>"
modules/video_models/video_run.py
+9 -3
View File
@@ -33,6 +33,14 @@ class VideoResult:
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]:
"""Return (selected, needs_load): a registry row when both names are given, or a synthesized
row describing the already-loaded pipeline when both are omitted."""
@@ -140,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,
modules/video_models/video_utils.py
+11
View File
@@ -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,16 @@ 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.
test/test-ltx-core.py
+443
View File
@@ -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)