indigo/LTX-2
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Automated PR - 2026-05-28

Browse Source
This commit is contained in:
github-actions[bot]
2026-05-28 14:26:15 +00:00
parent 1799988521
commit 203d4842d4
45 changed files with 1969 additions and 668 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/ltx-pipelines/src/ltx_pipelines/utils/args.py
+146 -26
View File
@@ -1,9 +1,11 @@
import argparse
import json
from collections.abc import Sequence
from pathlib import Path
from typing import NamedTuple
from typing import Any, NamedTuple
from ltx_core.loader import LTXV_LORA_COMFY_RENAMING_MAP, LoraPathStrengthAndSDOps
from ltx_core.model.transformer.compiling import CompilationConfig
from ltx_core.quantization import QuantizationPolicy
from ltx_pipelines.utils.constants import (
DEFAULT_IMAGE_CRF,
@@ -13,6 +15,7 @@ from ltx_pipelines.utils.constants import (
LTX_2_3_PARAMS,
PipelineParams,
)
from ltx_pipelines.utils.quantization_factory import QuantizationKind
from ltx_pipelines.utils.types import OffloadMode
@@ -32,7 +35,7 @@ class VideoConditioningAction(argparse.Action):
option_string: str | None = None, # noqa: ARG002
) -> None:
path, strength_str = values
resolved_path = resolve_path(path)
resolved_path = resolve_existing_path(path)
strength = float(strength_str)
current = getattr(namespace, self.dest) or []
current.append((resolved_path, strength))
@@ -58,7 +61,7 @@ class VideoMaskConditioningAction(argparse.Action):
msg = f"{option_string} requires exactly 2 arguments (MASK_PATH STRENGTH), got {len(values)}"
raise argparse.ArgumentError(self, msg)
mask_path = resolve_path(values[0])
mask_path = resolve_existing_path(values[0])
strength = float(values[1])
setattr(namespace, self.dest, (mask_path, strength))
@@ -76,7 +79,7 @@ class ImageAction(argparse.Action):
raise argparse.ArgumentError(self, msg)
conditioning = ImageConditioningInput(
path=resolve_path(values[0]),
path=resolve_existing_path(values[0]),
frame_idx=int(values[1]),
strength=float(values[2]),
crf=int(values[3]) if len(values) > 3 else DEFAULT_IMAGE_CRF,
@@ -101,7 +104,7 @@ class LoraAction(argparse.Action):
path = values[0]
strength_str = values[1] if len(values) > 1 else str(DEFAULT_LORA_STRENGTH)
resolved_path = resolve_path(path)
resolved_path = resolve_existing_path(path)
strength = float(strength_str)
current = getattr(namespace, self.dest) or []
@@ -109,11 +112,118 @@ class LoraAction(argparse.Action):
setattr(namespace, self.dest, current)
class CompileAction(argparse.Action):
"""Parse ``--compile [KEY=VALUE ...]`` into a :class:`CompilationConfig`.
The flag is absent -> ``args.compile`` stays at its default (``None``).
The flag is passed alone -> ``CompilationConfig()`` (vanilla torch defaults).
The flag is passed with args -> ``CompilationConfig`` with the given fields overridden.
Errors (unknown key, malformed value, duplicate key, empty value) raise
:class:`argparse.ArgumentError` so argparse formats them as friendly CLI
messages rather than uncaught tracebacks.
"""
_ALLOWED_KEYS = frozenset({"mode", "backend", "fullgraph", "dynamic", "inductor_config", "dynamo_config"})
def __call__(
self,
parser: argparse.ArgumentParser, # noqa: ARG002
namespace: argparse.Namespace,
values: list[str],
option_string: str | None = None, # noqa: ARG002
) -> None:
overrides: dict[str, object] = {}
for item in values:
if "=" not in item:
raise argparse.ArgumentError(self, f"expects KEY=VALUE pairs, got: {item!r}")
key, _, raw = item.partition("=")
key = key.strip()
if key not in self._ALLOWED_KEYS:
raise argparse.ArgumentError(
self,
f"{key!r} is not a CompilationConfig field; valid keys: {sorted(self._ALLOWED_KEYS)}",
)
if key in overrides:
raise argparse.ArgumentError(self, f"{key} given more than once")
if key == "mode":
overrides[key] = self._parse_mode(raw)
elif key == "backend":
overrides[key] = self._parse_non_empty(key, raw)
elif key == "fullgraph":
overrides[key] = self._parse_bool(key, raw)
elif key == "dynamic":
overrides[key] = self._parse_dynamic(raw)
elif key in ("inductor_config", "dynamo_config"):
overrides[key] = self._parse_json_dict(key, raw)
setattr(namespace, self.dest, CompilationConfig(**overrides))
def _parse_mode(self, raw: str) -> str | None:
stripped = raw.strip()
if not stripped:
raise argparse.ArgumentError(self, "mode=... value cannot be empty (use mode=none to clear)")
if stripped.lower() == "none":
return None
return stripped
def _parse_non_empty(self, key: str, raw: str) -> str:
stripped = raw.strip()
if not stripped:
raise argparse.ArgumentError(self, f"{key}=... value cannot be empty")
return stripped
def _parse_bool(self, key: str, raw: str) -> bool:
normalized = raw.strip().lower()
if normalized in ("true", "1"):
return True
if normalized in ("false", "0"):
return False
raise argparse.ArgumentError(self, f"{key}=... must be true or false; got {raw!r}")
def _parse_dynamic(self, raw: str) -> bool | None:
normalized = raw.strip().lower()
if normalized in ("auto", "none"):
return None
if normalized in ("true", "1"):
return True
if normalized in ("false", "0"):
return False
raise argparse.ArgumentError(self, f"dynamic=... must be auto/true/false; got {raw!r}")
def _parse_json_dict(self, key: str, raw: str) -> dict[str, Any]:
# Inline JSON object starts with '{'; otherwise treat the value as a path to a JSON file.
stripped = raw.strip()
if not stripped:
raise argparse.ArgumentError(self, f"{key}=... value cannot be empty")
if stripped.startswith("{"):
source = stripped
else:
path = Path(stripped).expanduser()
if not path.is_file():
raise argparse.ArgumentError(
self, f"{key}=... must be a JSON object or a path to a JSON file; got {raw!r}"
)
source = path.read_text()
try:
value = json.loads(source)
except json.JSONDecodeError as e:
raise argparse.ArgumentError(self, f"{key}=... must be a JSON object; got {raw!r} ({e.msg})") from None
if not isinstance(value, dict):
raise argparse.ArgumentError(self, f"{key}=... must decode to a JSON object; got {type(value).__name__}")
return value
def resolve_path(path: str) -> str:
return str(Path(path).expanduser().resolve().as_posix())
QUANTIZATION_POLICIES = ("fp8-cast", "fp8-scaled-mm")
def resolve_existing_path(path: str) -> str:
"""Resolve *path* and verify it exists."""
resolved = resolve_path(path)
if not Path(resolved).exists():
raise argparse.ArgumentError(None, f"Path not found: {resolved}")
return resolved
QUANTIZATION_POLICIES = tuple(k.value for k in QuantizationKind)
def _resolve_quantization(namespace: argparse.Namespace) -> None:
@@ -123,16 +233,14 @@ def _resolve_quantization(namespace: argparse.Namespace) -> None:
name = getattr(namespace, "quantization", None)
if name is None or isinstance(name, QuantizationPolicy):
return
if name == "fp8-cast":
namespace.quantization = QuantizationPolicy.fp8_cast()
try:
kind = QuantizationKind(name)
except ValueError:
return
if name == "fp8-scaled-mm":
ckpt = getattr(namespace, "checkpoint_path", None) or getattr(namespace, "distilled_checkpoint_path", None)
if ckpt is None:
raise SystemExit(
"--quantization fp8-scaled-mm requires --checkpoint-path (or --distilled-checkpoint-path)."
)
namespace.quantization = QuantizationPolicy.fp8_scaled_mm(ckpt)
ckpt = getattr(namespace, "checkpoint_path", None) or getattr(namespace, "distilled_checkpoint_path", None)
if ckpt is None:
raise SystemExit(f"--quantization {kind.value} requires --checkpoint-path (or --distilled-checkpoint-path).")
namespace.quantization = kind.to_policy(checkpoint_path=ckpt)
class _PipelineArgumentParser(argparse.ArgumentParser):
@@ -150,7 +258,7 @@ def detect_checkpoint_path(distilled: bool = False) -> str:
"""Pre-parse argv to extract the checkpoint path before building the full parser."""
pre = argparse.ArgumentParser(add_help=False)
flag = "--distilled-checkpoint-path" if distilled else "--checkpoint-path"
pre.add_argument(flag, type=resolve_path, required=True)
pre.add_argument(flag, type=resolve_existing_path, required=True)
known, _ = pre.parse_known_args()
return known.distilled_checkpoint_path if distilled else known.checkpoint_path
@@ -163,14 +271,14 @@ def basic_arg_parser(
if distilled:
parser.add_argument(
"--distilled-checkpoint-path",
type=resolve_path,
type=resolve_existing_path,
required=True,
help="Path to LTX-2 distilled model checkpoint (.safetensors file).",
)
else:
parser.add_argument(
"--checkpoint-path",
type=resolve_path,
type=resolve_existing_path,
required=True,
help="Path to LTX-2 model checkpoint (.safetensors file).",
)
@@ -185,7 +293,7 @@ def basic_arg_parser(
)
parser.add_argument(
"--gemma-root",
type=resolve_path,
type=resolve_existing_path,
required=True,
help="Path to the root directory containing the Gemma text encoder model files.",
)
@@ -276,8 +384,20 @@ def basic_arg_parser(
)
parser.add_argument(
"--compile",
action="store_true",
help="Enable torch.compile for transformer blocks to optimize performance.",
nargs="*",
action=CompileAction,
default=None,
metavar="KEY=VALUE",
help=(
"Enable torch.compile for transformer blocks. Pass alone for defaults, "
"or with KEY=VALUE overrides for any CompilationConfig field. "
"Keys: mode, backend, fullgraph, dynamic, inductor_config, dynamo_config. "
"inductor_config/dynamo_config take JSON objects (inline or a path to a .json file) "
"that fully replace the defaults. "
"Examples: --compile or --compile mode=reduce-overhead or "
"--compile mode=reduce-overhead fullgraph=true backend=eager or "
"--compile inductor_config='{\"max_autotune\": true}'"
),
)
return parser
@@ -340,7 +460,7 @@ def video_editing_arg_parser(
(no height/width/num-frames; resolution comes from input video). Default is distilled checkpoint only.
"""
parser = basic_arg_parser(distilled=distilled)
parser.add_argument("--video-path", type=resolve_path, required=True, help="Path to the source video.")
parser.add_argument("--video-path", type=resolve_existing_path, required=True, help="Path to the source video.")
parser.add_argument("--start-time", type=float, required=True, help="Start time of the region to regenerate (s).")
parser.add_argument("--end-time", type=float, required=True, help="End time of the region to regenerate (s).")
return parser
@@ -462,7 +582,7 @@ def default_1_stage_arg_parser(params: PipelineParams = LTX_2_3_PARAMS) -> argpa
default=video_guider.skip_step,
help=(
"Video skip step N controls periodic skipping during the video diffusion process: "
"only steps where step_index % (N + 1) == 0 are processed, all others are skipped "
"only steps where step_index %% (N + 1) == 0 are processed, all others are skipped "
f"(e.g., 0 = no skipping; 1 = skip every other step; 2 = skip 2 of every 3 steps; "
f"default: {video_guider.skip_step})."
),
@@ -522,7 +642,7 @@ def default_1_stage_arg_parser(params: PipelineParams = LTX_2_3_PARAMS) -> argpa
default=audio_guider.skip_step,
help=(
"Audio skip step N controls periodic skipping during the audio diffusion process: "
"only steps where step_index % (N + 1) == 0 are processed, all others are skipped "
"only steps where step_index %% (N + 1) == 0 are processed, all others are skipped "
f"(e.g., 0 = no skipping; 1 = skip every other step; 2 = skip 2 of every 3 steps; "
f"default: {audio_guider.skip_step})."
),
@@ -562,7 +682,7 @@ def default_2_stage_arg_parser(params: PipelineParams = LTX_2_3_PARAMS) -> argpa
)
parser.add_argument(
"--spatial-upsampler-path",
type=resolve_path,
type=resolve_existing_path,
required=True,
help=(
"Path to the spatial upsampler model used to increase the resolution "
@@ -605,7 +725,7 @@ def default_2_stage_distilled_arg_parser(params: PipelineParams = LTX_2_3_PARAMS
)
parser.add_argument(
"--spatial-upsampler-path",
type=resolve_path,
type=resolve_existing_path,
required=True,
help=(
"Path to the spatial upsampler model used to increase the resolution "