| """Extract six mode-specific workflow templates from the master LTX 2.3 All-In-One workflow. |
| |
| Each ComfyUI group whose title starts with a number (e.g. "01 Text to Video") becomes |
| a mode template containing only that group's nodes plus shared scaffolding (Models, |
| Lora, Setting, Prompt, Load Audio/Image/Video, Output groups). |
| |
| Group title -> output filename mapping: |
| 01 -> t2v.json |
| 02 -> a2v.json |
| 03 -> i2v.json |
| 04 -> lipsync.json |
| 05 -> keyframe.json |
| 06 -> style.json |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import pathlib |
| import sys |
| from collections.abc import Iterable |
|
|
| GROUP_TO_FILENAME: dict[str, str] = { |
| "01": "t2v.json", |
| "02": "a2v.json", |
| "03": "i2v.json", |
| "04": "lipsync.json", |
| "05": "keyframe.json", |
| "06": "style.json", |
| } |
|
|
| SHARED_GROUP_PREFIXES: tuple[str, ...] = ( |
| "Models", |
| "Lora", |
| "Setting", |
| "Prompt", |
| "Load Audio", |
| "Load Image", |
| "Load Images", |
| "Load Video", |
| "Output", |
| ) |
|
|
|
|
| def _node_in_group(node: dict, group: dict) -> bool: |
| """Test whether a node's position lies inside a group's bounding box.""" |
| if "pos" not in node or "bounding" not in group: |
| return False |
| nx, ny = node["pos"][0], node["pos"][1] |
| gx, gy, gw, gh = group["bounding"] |
| return (gx <= nx <= gx + gw) and (gy <= ny <= gy + gh) |
|
|
|
|
| def _select_groups(master: dict, mode_prefix: str) -> list[dict]: |
| """Pick the mode group plus all shared groups.""" |
| selected: list[dict] = [] |
| for g in master.get("groups", []): |
| title = (g.get("title") or "").strip() |
| if title.startswith(mode_prefix + " "): |
| selected.append(g) |
| elif any(title.startswith(p) for p in SHARED_GROUP_PREFIXES): |
| selected.append(g) |
| return selected |
|
|
|
|
| def _collect_nodes(master: dict, groups: Iterable[dict]) -> list[dict]: |
| """Return all nodes lying inside any of the given groups.""" |
| groups_list = list(groups) |
| keep: list[dict] = [] |
| for node in master.get("nodes", []): |
| if any(_node_in_group(node, g) for g in groups_list): |
| keep.append(node) |
| return keep |
|
|
|
|
| def _collect_links(master: dict, kept_node_ids: set[int]) -> list[list]: |
| """Keep only links where both endpoints are in the surviving node set.""" |
| return [ |
| link |
| for link in master.get("links", []) |
| |
| if link[1] in kept_node_ids and link[3] in kept_node_ids |
| ] |
|
|
|
|
| def extract_mode(master: dict, mode_prefix: str) -> dict: |
| """Build a focused workflow JSON for the given mode group prefix.""" |
| groups = _select_groups(master, mode_prefix) |
| nodes = _collect_nodes(master, groups) |
| kept_ids = {n["id"] for n in nodes} |
| links = _collect_links(master, kept_ids) |
|
|
| return { |
| "id": f"ltx23-aio-{mode_prefix}", |
| "revision": 0, |
| "last_node_id": max(kept_ids, default=0), |
| "last_link_id": max((link[0] for link in links), default=0), |
| "nodes": nodes, |
| "links": links, |
| "groups": groups, |
| "definitions": master.get("definitions", {}), |
| "config": master.get("config", {}), |
| "extra": master.get("extra", {}), |
| "version": master.get("version", 0.4), |
| } |
|
|
|
|
| def main(argv: list[str] | None = None) -> int: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--master", type=pathlib.Path, required=True) |
| parser.add_argument("--out", type=pathlib.Path, required=True) |
| args = parser.parse_args(argv) |
|
|
| master = json.loads(args.master.read_text()) |
| args.out.mkdir(parents=True, exist_ok=True) |
|
|
| for prefix, filename in GROUP_TO_FILENAME.items(): |
| wf = extract_mode(master, prefix) |
| out_path = args.out / filename |
| out_path.write_text(json.dumps(wf, indent=2)) |
| print(f" -> wrote {out_path} ({len(wf['nodes'])} nodes, {len(wf['links'])} links)") |
|
|
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| sys.exit(main()) |
|
|
