diff --git a/modules/api/video.py b/modules/api/video.py index ff2345214..9afc66385 100644 --- a/modules/api/video.py +++ b/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: diff --git a/modules/ltx/ltx_ui.py b/modules/ltx/ltx_ui.py index 11f0d5cba..bd6cd709f 100644 --- a/modules/ltx/ltx_ui.py +++ b/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, diff --git a/modules/ltx/ltx_util.py b/modules/ltx/ltx_util.py index 914c86834..68e25f86a 100644 --- a/modules/ltx/ltx_util.py +++ b/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: diff --git a/modules/video_models/models_def.py b/modules/video_models/models_def.py index b8872fdab..5edeae458 100644 --- a/modules/video_models/models_def.py +++ b/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') diff --git a/modules/video_models/video_overrides.py b/modules/video_models/video_overrides.py index 1964bdc59..f0e760896 100644 --- a/modules/video_models/video_overrides.py +++ b/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) diff --git a/modules/video_models/video_run.py b/modules/video_models/video_run.py index 453e6738a..5f36ea8d0 100644 --- a/modules/video_models/video_run.py +++ b/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}"') diff --git a/modules/video_models/video_ui.py b/modules/video_models/video_ui.py index 894082314..c67549da1 100644 --- a/modules/video_models/video_ui.py +++ b/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(): diff --git a/test/test-video-registry.py b/test/test-video-registry.py new file mode 100644 index 000000000..b2ca6f1f8 --- /dev/null +++ b/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)