Merge pull request #5037 from vladmandic/feat/video-dispatch-mode

Feat/video dispatch mode

modules/api/video.py

@@ -72,27 +72,14 @@ class ItemVideoModel(BaseModel):
     name: str = Field(title="Name", description="Model name; pass together with engine to select it")
     repo: str = Field(default="", title="Repo", description="Model repository or path")
     url: str = Field(default="", title="URL", description="Model information page")
-    mode: str = Field(title="Mode", description="Input mode: workflow, t2v, i2v, flf2v, vace, or animate")
+    mode: str = Field(title="Mode", description="Input mode: workflow, t2v, i2v, flf2v, vace, animate, condition, or unknown; condition models accept conditioning the generic path does not wire and run as text to video here")
     workflow: str | None = Field(default=None, title="Workflow", description="Modular workflow name when the model dispatches on inputs; ref2va conditions on references and ignores the keyframe images")
     base: bool = Field(default=False, title="Base", description="Also listed in the base checkpoint dropdown")
     loaded: bool = Field(default=False, title="Loaded", description="Currently loaded through the video registry")
 
 
 def model_mode(m: models_def.Model) -> str:
-    # mirrors the dispatch order in video_run.run: workflow models route on inputs, the rest on name markers
-    if m.workflow is not None:
-        return 'workflow'
-    if 'T2V' in m.name:
-        return 't2v'
-    if 'I2V' in m.name:
-        return 'i2v'
-    if 'FLF2V' in m.name:
-        return 'flf2v'
-    if 'VACE' in m.name:
-        return 'vace'
-    if 'Animate' in m.name:
-        return 'animate'
-    return 't2v'
+    return models_def.dispatch_mode(m)
 
 
 class APIVideo:

modules/ltx/ltx_ui.py

@@ -128,9 +128,9 @@ def create_ui(prompt, negative, styles, overrides, script_inputs, mp4_fps, mp4_i
 
 
     def load_model(model_name: str):
-        ltx_util.load_model('LTX Video', model_name)
+        return ltx_util.load_model('LTX Video', model_name)
 
-    btn_load.click(fn=load_model, inputs=[model], outputs=[])
+    btn_load.click(fn=load_model, inputs=[model], outputs=[text])
 
     model.change(
         fn=_model_change,

modules/ltx/ltx_util.py

@@ -19,22 +19,27 @@ def get_frames(frames: int):
     return int(8 * (int(frames) // 8)) + 1
 
 
-def load_model(engine: str, model: str):
+def load_model(engine: str, model: str) -> str:
     if model is None or model == '' or model == 'None':
         shared.sd_model = None
-        return
-    t0 = time.time()
+        return 'Video model unloaded'
     from modules.video_models import models_def, video_load
-    selected: models_def.Model = [m for m in models_def.models[engine] if m.name == model][0]
+    selected = models_def.find(engine, model)
+    if selected is None: # the dropdown lists the separators it groups models under, and they name no model
+        msg = f'Video model not loaded: engine="{engine}" model="{model}"'
+        log.warning(msg)
+        return msg
+    t0 = time.time()
     # video_load owns the cache; pipe-class mismatch inside it invalidates the name-based hit
     # when Unload Models (or any external swap) silently replaced shared.sd_model.
     log.info(f'Load video: engine="{engine}" selected="{model}" {selected}')
-    video_load.load_model(selected)
+    msg = video_load.load_model(selected)
     t1 = time.time()
     shared.sd_model = sd_models.apply_balanced_offload(shared.sd_model)
     t2 = time.time()
     timer.process.add('load', t1 - t0)
     timer.process.add('offload', t2 - t1)
+    return msg or f'Video model loaded: {selected.name}'
 
 
 def upsample_pipe_stale(upsample_pipe, upsample_repo_id) -> bool:

modules/video_models/models_def.py

@@ -826,3 +826,61 @@ def pipeline_classes() -> set[str]:
             if row.custom is not None:
                 classes.add(row.custom)
     return classes
+
+
+NAME_MODES = ( # markers a row name carries to declare its inputs, in the order run() tests them
+    ('T2V', 't2v'),
+    ('I2V', 'i2v'),
+    ('FLF2V', 'flf2v'),
+    ('VACE', 'vace'),
+    ('Animate', 'animate'),
+)
+CLASS_MODES = { # the mode a pipeline class implies, for rows whose name declares nothing
+    'HunyuanVideoPipeline': 't2v',
+    'HunyuanVideo15Pipeline': 't2v',
+    'HunyuanVideoImageToVideoPipeline': 'i2v',
+    'HunyuanVideo15ImageToVideoPipeline': 'i2v',
+    'HunyuanSkyreelsImageToVideoPipeline': 'i2v',
+    'LTXPipeline': 't2v',
+    'LTX2Pipeline': 't2v',
+    'LTXImageToVideoPipeline': 'i2v',
+    'LTX2ImageToVideoPipeline': 'i2v',
+    'LTXConditionPipeline': 'condition',
+    'LTX2ConditionPipeline': 'condition',
+    'WanPipeline': 't2v',
+    'WanImageToVideoPipeline': 'i2v',
+    'WanVACEPipeline': 'vace',
+    'WanAnimatePipeline': 'animate',
+    'SkyReelsV2Pipeline': 't2v',
+    'SkyReelsV2DiffusionForcingPipeline': 't2v',
+    'SkyReelsV2ImageToVideoPipeline': 'i2v',
+    'SkyReelsV2DiffusionForcingImageToVideoPipeline': 'i2v',
+    'MochiPipeline': 't2v',
+    'LattePipeline': 't2v',
+    'AllegroPipeline': 't2v',
+    'CogVideoXPipeline': 't2v',
+    'CogVideoXImageToVideoPipeline': 'i2v',
+    'Cosmos2VideoToWorldPipeline': 'i2v',
+    'SanaVideoPipeline': 't2v',
+    'Kandinsky5T2VPipeline': 't2v',
+    'Kandinsky5I2VPipeline': 'i2v',
+    'MiniMaxH3ModularPipeline': 'workflow',
+    'GoogleVeoVideoPipeline': 't2v',
+}
+
+
+def dispatch_mode(row: Model) -> str:
+    """How a row's inputs are wired: workflow, t2v, i2v, flf2v, vace, animate, condition, or unknown.
+
+    Name markers are read before the pipeline class, since one class serves several modes: six
+    LTXConditionPipeline rows are named T2V or I2V and generate as such.
+    """
+    if row is None:
+        return 'unknown'
+    if row.workflow is not None:
+        return 'workflow'
+    for marker, mode in NAME_MODES:
+        if marker in (row.name or ''):
+            return mode
+    cls = row.repo_cls if isinstance(row.repo_cls, str) else getattr(row.repo_cls, '__name__', None)
+    return CLASS_MODES.get(cls or row.custom, 'unknown')

modules/video_models/video_overrides.py

@@ -91,6 +91,7 @@ def set_overrides(p: processing.StableDiffusionProcessingVideo, selected: Model)
     if 'LTX' in cls:
         p.task_args['width'] = 32 * (p.width // 32)
         p.task_args['height'] = 32 * (p.height // 32)
+        p.frames = 8 * (p.frames // 8) + 1 # same rule the ltx tab applies, so both paths request a length the pipe keeps
     # WAN
     if 'Wan' in cls:
         p.task_args['width'] = 16 * (p.width // 16)

modules/video_models/video_run.py

@@ -181,7 +181,8 @@ def run(selected: models_def.Model, *,
     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)
-    if getattr(selected, 'workflow', None) is not None:
+    mode = models_def.dispatch_mode(selected)
+    if mode == 'workflow':
         # modular workflows dispatch on which inputs are present; keyframes pass through
         # unresized since the pipeline defines its own canvas placement per anchor
         p.video_still = int(frames) <= 1
@@ -200,10 +201,10 @@ def run(selected: models_def.Model, *,
         elif int(mp4_fps) != 24:
             log.warning(f'Video: model="{selected.name}" fps={mp4_fps} model output is fixed at 24')
         log.debug(f'Video: op=modular workflow={selected.workflow} still={p.video_still} init={init_image} last={last_image} references={len(refs) if refs else 0}')
-    elif 'T2V' in selected.name:
+    elif mode == 't2v':
         if init_image is not None:
             log.warning('Video: op=T2V init image not supported')
-    elif 'I2V' in selected.name:
+    elif mode == 'i2v':
         if init_image is None:
             raise VideoError('No input image provided. Please upload or select an image.', 400)
         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')
@@ -214,7 +215,7 @@ def run(selected: models_def.Model, *,
             log.warning(f'Video: op=I2V model="{selected.name}" last frame not supported, ignoring')
         else:
             log.debug(f'Video: op=I2V init={init_image} resized={p.task_args["image"]}')
-    elif 'FLF2V' in selected.name:
+    elif mode == 'flf2v':
         if init_image is None:
             raise VideoError('No input image provided. Please upload or select an image.', 400)
         if last_image is None:
@@ -222,11 +223,11 @@ def run(selected: models_def.Model, *,
         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')
         p.task_args['last_image'] = images.resize_image(resize_mode=2, im=last_image, width=p.width, height=p.height, upscaler_name=None, output_type='pil')
         log.debug(f'Video: op=FLF2V init={init_image} last={last_image} resized={p.task_args["image"]}')
-    elif 'VACE' in selected.name:
+    elif mode == 'vace':
         if init_image is not None:
             p.task_args['reference_images'] = [images.resize_image(resize_mode=2, im=init_image, width=p.width, height=p.height, upscaler_name=None, output_type='pil')]
             log.debug(f'Video: op=VACE reference={init_image} resized={p.task_args["reference_images"]}')
-    elif 'Animate' in selected.name:
+    elif mode == 'animate':
         if init_image is None:
             raise VideoError('No input image provided. Please upload or select an image.', 400)
         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')
@@ -234,6 +235,11 @@ def run(selected: models_def.Model, *,
         p.task_args['pose_video'] = [] # input pose video to condition the generation on. must be a list of PIL images.
         p.task_args['face_video'] = [] # input face video to condition the generation on. must be a list of PIL images.
         log.debug(f'Video: op=Animate init={p.task_args["image"]} pose={p.task_args["pose_video"]} face={p.task_args["face_video"]}')
+    elif mode == 'condition':
+        # the conditioning inputs these models accept are wired on the ltx tab, not here
+        log.warning(f'Video: op=condition model="{selected.name}" conditioning not supported here, running text to video')
+        if init_image is not None:
+            log.warning(f'Video: op=condition model="{selected.name}" init image not supported, ignoring')
     else:
         log.warning(f'Video: unknown model type "{selected.name}"')
 

modules/video_models/video_ui.py

@@ -25,17 +25,12 @@ def engine_change(engine):
 
 
 def get_selected(engine, model):
-    found = [model.name for model in models_def.models.get(engine, [])]
-    if len(models_def.models[engine]) > 0 and len(found) > 0:
-        selected = [m for m in models_def.models[engine] if m.name == model][0]
-        return selected
-    return None
+    return models_def.find(engine, model)
 
 
 def model_change(engine, model):
     debug(f'Video change: engine="{engine}" model="{model}"')
-    found = [model.name for model in models_def.models.get(engine, [])]
-    selected = [m for m in models_def.models[engine] if m.name == model][0] if len(found) > 0 else None
+    selected = get_selected(engine, model)
     url = video_utils.get_url(selected.url if selected else None)
     return url
 
@@ -43,19 +38,21 @@ def model_change(engine, model):
 def model_load(engine, model):
     debug(f'Load video: engine="{engine}" model="{model}"')
     selected = get_selected(engine, model)
-    yield f'Video model loading: {selected.name}'
-    if selected:
-        if 'None' in selected.name:
+    if selected is None: # the dropdown lists the separators it groups models under, and they name no model
+        if model and model.startswith('─'):
+            msg = 'Video model not loaded: dropdown separator selected'
+        elif model in (None, '', 'None'):
             sd_models.unload_model_weights()
             msg = 'Video model unloaded'
         else:
-            from modules.video_models import video_load
-            msg = video_load.load_model(selected)
-    else:
-        sd_models.unload_model_weights()
-        msg = 'Video model unloaded'
+            msg = f'Video model not found: engine="{engine}" model="{model}"'
+        log.warning(msg)
+        yield msg
+        return
+    yield f'Video model loading: {selected.name}'
+    from modules.video_models import video_load
+    msg = video_load.load_model(selected)
     yield msg
-    return msg
 
 
 def create_ui_outputs():

test/test-video-registry.py

@@ -0,0 +1,315 @@
+#!/usr/bin/env python
+"""
+Offline unit tests for the video model registry in modules.video_models.models_def.
+
+The registry answers two questions about a row: whether it names a loadable model, and how a
+runner should wire its inputs. Both were previously recovered from display-name substrings at
+each call site, which drifted.
+
+Covers:
+
+- the sentinel contract: the None placeholder and the dropdown separators name no model, and
+  every accessor excludes them
+- row uniqueness, so a duplicated entry cannot reach the dropdown twice
+- ``dispatch_mode`` totality: every registered row classifies, so a new row that declares
+  neither a name marker nor a mapped pipeline class fails here rather than generating as t2v
+- ``dispatch_mode`` equivalence against the ladder it replaced, for every row the ladder
+  classified
+- the eight condition rows the ladder did not classify, and the six condition-class rows whose
+  names do declare a mode
+
+No running server required.
+
+Usage:
+    python test/test-video-registry.py
+"""
+
+import os
+import sys
+
+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 models_def          # pylint: disable=wrong-import-position
+
+
+results: dict[str, dict] = {}
+
+
+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 loadable_rows():
+    for engine, rows in models_def.models.items():
+        for row in rows:
+            if models_def.is_model(row):
+                yield engine, row
+
+
+def sentinel_rows():
+    for engine, rows in models_def.models.items():
+        for row in rows:
+            if not models_def.is_model(row):
+                yield engine, row
+
+
+def old_ladder(row):
+    """The name-marker ladder dispatch_mode replaced, as the oracle for equivalence.
+
+    'unknown' stands for the branch in run() that warned and wired nothing; the api ladder
+    reported those same rows as t2v.
+    """
+    if row.workflow is not None:
+        return 'workflow'
+    if 'T2V' in row.name:
+        return 't2v'
+    if 'I2V' in row.name:
+        return 'i2v'
+    if 'FLF2V' in row.name:
+        return 'flf2v'
+    if 'VACE' in row.name:
+        return 'vace'
+    if 'Animate' in row.name:
+        return 'animate'
+    return 'unknown'
+
+
+# ============================================================
+# Sentinel contract and row identity
+# ============================================================
+
+def test_registry_is_populated():
+    assert len(models_def.models) > 0, 'registry failed to build'
+    assert sum(1 for _ in loadable_rows()) > 50, 'registry lost most of its rows'
+
+
+def test_sentinels_name_no_model():
+    for engine, row in sentinel_rows():
+        assert row.name == 'None' or row.name.startswith('─'), f'[{engine}] unexpected sentinel {row.name}'
+        assert row.repo is None, f'[{engine}] sentinel "{row.name}" carries a repo'
+
+
+def test_accessors_exclude_sentinels():
+    for engine in models_def.models:
+        names = models_def.model_names(engine)
+        for name in names:
+            assert name != 'None' and not name.startswith('─'), f'[{engine}] sentinel "{name}" listed as a model'
+    for engine in models_def.engines():
+        assert models_def.model_names(engine), f'[{engine}] listed as an engine with no models'
+
+
+def test_find_rejects_sentinels():
+    for engine, row in sentinel_rows():
+        assert models_def.find(engine, row.name) is None, f'[{engine}] find resolved sentinel "{row.name}"'
+
+
+def test_find_is_case_insensitive():
+    engine, row = next(iter(loadable_rows()))
+    assert models_def.find(engine, row.name) is row
+    assert models_def.find(engine.lower(), row.name.lower()) is row
+    assert models_def.find(engine.upper(), row.name.upper()) is row
+
+
+def test_find_rejects_unknown_names():
+    engine, _row = next(iter(loadable_rows()))
+    assert models_def.find(engine, 'no such model') is None
+    assert models_def.find('no such engine', 'no such model') is None
+    assert models_def.find(engine, None) is None
+
+
+def test_rows_are_unique_within_an_engine():
+    seen = {}
+    for engine, row in loadable_rows():
+        key = (engine, row.name.lower())
+        assert key not in seen, f'[{engine}] duplicate row "{row.name}"'
+        seen[key] = row
+
+
+# ============================================================
+# Mode derivation
+# ============================================================
+
+def test_every_row_classifies():
+    """The extensibility gate: a row declaring neither a marker nor a mapped class fails here."""
+    unknown = [f'[{engine}] {row.name}' for engine, row in loadable_rows() if models_def.dispatch_mode(row) == 'unknown']
+    assert not unknown, f'rows with no mode: {unknown}'
+
+
+def test_mode_matches_the_ladder_it_replaced():
+    """Every row the old ladder classified keeps its answer; only its blind spot changes."""
+    changed = []
+    for engine, row in loadable_rows():
+        old = old_ladder(row)
+        new = models_def.dispatch_mode(row)
+        if old != 'unknown' and old != new:
+            changed.append(f'[{engine}] {row.name}: {old} -> {new}')
+    assert not changed, f'mode changed on rows the ladder already handled: {changed}'
+
+
+def test_the_ladder_blind_spot_is_the_condition_rows():
+    unclassified = [row for _engine, row in loadable_rows() if old_ladder(row) == 'unknown']
+    assert len(unclassified) == 8, f'expected 8 rows the ladder missed, found {len(unclassified)}'
+    for row in unclassified:
+        assert models_def.dispatch_mode(row) == 'condition', f'"{row.name}" resolved as {models_def.dispatch_mode(row)}'
+
+
+def test_named_modes_win_over_the_pipeline_class():
+    """Six LTXConditionPipeline rows are named T2V or I2V and generate as such."""
+    checked = 0
+    for _engine, row in loadable_rows():
+        cls = row.repo_cls if isinstance(row.repo_cls, str) else getattr(row.repo_cls, '__name__', None)
+        if cls not in ('LTXConditionPipeline', 'LTX2ConditionPipeline'):
+            continue
+        if 'T2V' in row.name:
+            assert models_def.dispatch_mode(row) == 't2v', f'"{row.name}" lost its declared mode'
+            checked += 1
+        elif 'I2V' in row.name:
+            assert models_def.dispatch_mode(row) == 'i2v', f'"{row.name}" lost its declared mode'
+            checked += 1
+    assert checked == 6, f'expected 6 condition-class rows declaring a mode, found {checked}'
+
+
+def test_workflow_rows_report_workflow():
+    rows = [row for _engine, row in loadable_rows() if row.workflow is not None]
+    assert rows, 'registry carries no workflow rows'
+    for row in rows:
+        assert models_def.dispatch_mode(row) == 'workflow', f'"{row.name}" resolved as {models_def.dispatch_mode(row)}'
+
+
+def test_flf2v_row_reports_flf2v():
+    rows = [row for _engine, row in loadable_rows() if 'FLF2V' in row.name]
+    assert rows, 'registry carries no flf2v row'
+    for row in rows:
+        assert models_def.dispatch_mode(row) == 'flf2v', f'"{row.name}" resolved as {models_def.dispatch_mode(row)}'
+
+
+def test_mode_resolves_from_a_class_object():
+    """resolve_model synthesizes rows whose repo_cls is a class, not the registry's string."""
+    class WanPipeline: # pylint: disable=too-few-public-methods
+        pass
+    row = models_def.Model(name='local folder with no markers', repo_cls=WanPipeline)
+    assert models_def.dispatch_mode(row) == 't2v'
+
+
+def test_mode_resolves_a_custom_pipeline():
+    row = models_def.Model(name='local folder with no markers', custom='GoogleVeoVideoPipeline')
+    assert models_def.dispatch_mode(row) == 't2v'
+
+
+def test_missing_row_is_unknown():
+    assert models_def.dispatch_mode(None) == 'unknown'
+    assert models_def.dispatch_mode(models_def.Model(name='unregistered')) == 'unknown'
+
+
+def test_class_table_has_no_stale_entries():
+    """Every mapped class is one the registry actually uses, so the table cannot rot unnoticed."""
+    registered = models_def.pipeline_classes()
+    stale = [cls for cls in models_def.CLASS_MODES if cls not in registered]
+    assert not stale, f'class table names pipelines the registry does not carry: {stale}'
+
+
+def run_all():
+    log.warning('=== sentinels and identity ===')
+    cat = category('registry')
+    for fn in [
+        test_registry_is_populated,
+        test_sentinels_name_no_model,
+        test_accessors_exclude_sentinels,
+        test_find_rejects_sentinels,
+        test_find_is_case_insensitive,
+        test_find_rejects_unknown_names,
+        test_rows_are_unique_within_an_engine,
+    ]:
+        run_test(cat, fn)
+
+    log.warning('=== mode derivation ===')
+    cat = category('mode')
+    for fn in [
+        test_every_row_classifies,
+        test_mode_matches_the_ladder_it_replaced,
+        test_the_ladder_blind_spot_is_the_condition_rows,
+        test_named_modes_win_over_the_pipeline_class,
+        test_workflow_rows_report_workflow,
+        test_flf2v_row_reports_flf2v,
+        test_mode_resolves_from_a_class_object,
+        test_mode_resolves_a_custom_pipeline,
+        test_missing_row_is_unknown,
+        test_class_table_has_no_stale_entries,
+    ]:
+        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)