diff --git a/modules/ltx/ltx_process.py b/modules/ltx/ltx_process.py
index 5d8836f6e..c05989e37 100644
--- a/modules/ltx/ltx_process.py
+++ b/modules/ltx/ltx_process.py
@@ -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""
+
+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""
diff --git a/modules/ltx/ltx_util.py b/modules/ltx/ltx_util.py
index 68e25f86a..74f07f52e 100644
--- a/modules/ltx/ltx_util.py
+++ b/modules/ltx/ltx_util.py
@@ -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))
diff --git a/modules/video_models/video_run.py b/modules/video_models/video_run.py
index 5f36ea8d0..6c80412fc 100644
--- a/modules/video_models/video_run.py
+++ b/modules/video_models/video_run.py
@@ -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,
diff --git a/modules/video_models/video_utils.py b/modules/video_models/video_utils.py
index 1444d4782..82cc41eed 100644
--- a/modules/video_models/video_utils.py
+++ b/modules/video_models/video_utils.py
@@ -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()
diff --git a/test/test-ltx-core.py b/test/test-ltx-core.py
new file mode 100644
index 000000000..a26d5980b
--- /dev/null
+++ b/test/test-ltx-core.py
@@ -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)