Shared denoise execution primitives for LTX pipelines.
module-attribute ¶
LTXDenoiseStep = Callable[
[int, torch.Tensor, LTXAVState], LTXAVState
]
dataclass ¶
Run the one shared LTX denoise loop.
Prediction and scheduler math remain injectable so structural refactors do not change the existing LTX2/LTX2.3 numerical paths. Guidance will replace that step policy independently.
staticmethod ¶
run(
pipeline: LTXDenoisePipeline,
state: LTXAVState,
timesteps: Iterable[Tensor],
step: LTXDenoiseStep,
) -> LTXAVState
dataclass ¶
Immutable metadata and schedulers for one LTX denoise phase.
Prepare and execute one LTX phase without owning model modules.
staticmethod ¶
run(
pipeline: Any,
req: DiffusionRequestBatch,
request_inputs: LTXRequestInputs,
*,
noise_scale: float,
sigmas: list[float] | None,
timesteps: list[int] | None,
attention_kwargs: dict[str, Any] | None,
phase_recipe: LTXPhaseRecipe,
image: Any | None = None,
prompt_context: LTXPromptContext | None = None,
) -> LTXPhaseResult
dataclass ¶
Denoised AV latents and the context used to produce them.
class-attribute instance-attribute ¶
Expose the shared LTX Euler update through the distributed scheduler API.
build_transformer_kwargs(
pipeline: Any,
forward_ctx: LTXForwardContext,
denoise_ctx: LTXDenoiseContext,
*,
hidden_states: Tensor,
audio_hidden_states: Tensor,
encoder_hidden_states: Tensor,
audio_encoder_hidden_states: Tensor,
encoder_attention_mask: Tensor | None,
audio_encoder_attention_mask: Tensor | None,
ts: Tensor,
attention_kwargs: dict[str, Any] | None = None,
) -> dict[str, Any]
calculate_shift(
image_seq_len: int,
base_seq_len: int = 256,
max_seq_len: int = 4096,
base_shift: float = 0.5,
max_shift: float = 1.15,
) -> float
prepare_rope_coords_stage(
pipeline: Any,
forward_ctx: LTXForwardContext,
latents: Tensor,
audio_latents: Tensor,
) -> tuple[Tensor, Tensor]
prepare_scheduler_stage(
pipeline: Any,
request_inputs: LTXRequestInputs,
*,
device: device,
sigmas: list[float] | None,
timesteps: list[int] | None,
latent_num_frames: int,
latent_height: int,
latent_width: int,
use_official_sigma_schedule: bool,
image_conditioned: bool = False,
sampler: str = "euler",
generator: Generator | list[Generator] | None = None,
conditioning_mask: Tensor | None = None,
) -> tuple[Any, Any, Tensor]
step_denoised_latents(
pipeline: Any,
forward_ctx: LTXForwardContext,
denoise_ctx: LTXDenoiseContext,
noise_pred_video: Tensor,
noise_pred_audio: Tensor,
timestep: Tensor,
) -> tuple[Tensor, Tensor]