Skip to content

vllm_omni.diffusion.models.anima.pipeline_anima

ANIMA_VAE_SCALE_FACTOR module-attribute 

ANIMA_VAE_SCALE_FACTOR = 8

logger module-attribute 

logger = logging.getLogger(__name__)

AnimaPipeline

Bases: Module, DiffusionPipelineProfilerMixin, ProgressBarMixin

Native Anima model path.

The Anima transformer checkpoint is distributed as a single safetensors file. This loader builds the native pipeline components and runs the denoise transformer/text-conditioner directly.

current_timestep property 

current_timestep: Tensor | None

device instance-attribute 

device = device or get_local_device()

guidance_scale property 

guidance_scale: float

interrupt property 

interrupt: bool

num_timesteps property 

num_timesteps: int

od_config instance-attribute 

od_config = od_config

parallel_config instance-attribute 

parallel_config = od_config.parallel_config

supports_step_execution class-attribute instance-attribute 

supports_step_execution = False

weights_sources instance-attribute 

weights_sources: list[str] = []

condition_prompt_embeds 

condition_prompt_embeds(
    qwen_prompt_embeds: Tensor,
    qwen_attention_mask: Tensor,
    t5_input_ids: Tensor,
    t5_attention_mask: Tensor,
    device: device | None = None,
    conditioning_dtype: dtype | None = None,
    output_dtype: dtype | None = None,
) -> Tensor

decode_latents 

decode_latents(
    latents: Tensor, output_type: str = "pil"
) -> DiffusionOutput

Decode final latents.

diffuse 

diffuse(
    prompt_embeds: Tensor,
    negative_prompt_embeds: Tensor | None,
    latents: Tensor,
    padding_mask: Tensor,
    timesteps: Tensor,
    do_true_cfg: bool,
    true_cfg_scale: float,
) -> Tensor

encode_prompt 

encode_prompt(
    prompt: str | list[str],
    negative_prompt: str | list[str] | None = None,
    prepare_unconditional_embeds: bool = True,
    max_sequence_length: int = 512,
    device: device | None = None,
    dtype: dtype | None = None,
) -> dict[str, Tensor | None]

forward 

forward(
    req: DiffusionRequestBatch,
    prompt: str | list[str] | None = None,
    negative_prompt: str | list[str] | None = None,
    true_cfg_scale: float = 4.0,
    height: int | None = None,
    width: int | None = None,
    num_inference_steps: int = 50,
    sigmas: list[float] | None = None,
    guidance_scale: float = 4.0,
    num_images_per_prompt: int = 1,
    generator: Generator | list[Generator] | None = None,
    latents: Tensor | None = None,
    output_type: str | None = "pil",
    max_sequence_length: int = 512,
) -> list[DiffusionOutput]

load_weights 

load_weights(
    weights: Iterable[tuple[str, Tensor]] | None = None,
) -> set[str]

prepare_latents 

prepare_latents(
    batch_size: int,
    num_channels_latents: int,
    height: int,
    width: int,
    dtype: dtype | None = None,
    device: device | None = None,
    generator: Generator | list[Generator] | None = None,
    latents: Tensor | None = None,
) -> Tensor

prepare_timesteps 

prepare_timesteps(
    num_inference_steps: int,
    sigmas: list[float] | None = None,
    device: device | None = None,
) -> tuple[Tensor, int]

get_anima_post_process_func 

get_anima_post_process_func(
    od_config: OmniDiffusionConfig,
) -> Callable[[Tensor], Any]

retrieve_timesteps 

retrieve_timesteps(
    scheduler: Any,
    num_inference_steps: int | None = None,
    device: str | device | None = None,
    timesteps: list[int] | None = None,
    sigmas: list[float] | None = None,
    **kwargs: Any,
) -> tuple[Tensor, int]