methods/cache_strategy/DiCache/runtime.py

from __future__ import annotations

import types
from typing import List, Optional, Tuple, Union

import torch
import torch.amp as amp

from methods.cache_strategy.common import (
    DiCacheConfig,
    _maybe_concat_condition_mask,
    _maybe_embed_action,
    initialize_dicache_state,
    prepare_cache_runtime_model,
)

try:
    try:
        from cosmos_predict2.conditioner import DataType
    except ImportError:
        from cosmos_predict2._src.predict2.conditioner import DataType
except Exception:  # pragma: no cover - test fallback for minimal environments
    from enum import Enum

    class DataType(Enum):
        VIDEO = "video"
        IMAGE = "image"

try:
    try:
        from imaginaire.utils import log
    except ImportError:
        from cosmos_predict2._src.imaginaire.utils import log
except Exception:  # pragma: no cover - test fallback for minimal environments
    class _FallbackLog:
        @staticmethod
        def info(*args, **kwargs):
            pass

    log = _FallbackLog()


def dicache_mini_train_dit_forward(
    self,
    x_B_C_T_H_W: torch.Tensor,
    timesteps_B_T: torch.Tensor,
    crossattn_emb: torch.Tensor,
    fps: Optional[torch.Tensor] = None,
    padding_mask: Optional[torch.Tensor] = None,
    data_type: Optional[DataType] = DataType.VIDEO,
    intermediate_feature_ids: Optional[List[int]] = None,
    img_context_emb: Optional[torch.Tensor] = None,
    condition_video_input_mask_B_C_T_H_W: Optional[torch.Tensor] = None,
    **kwargs,
) -> Union[torch.Tensor, Tuple[torch.Tensor, List[torch.Tensor]]]:
    del intermediate_feature_ids
    assert isinstance(data_type, DataType), f"Expected DataType, got {type(data_type)}."

    if kwargs.get("timestep_scale") is None and hasattr(self, "timestep_scale"):
        timesteps_B_T = timesteps_B_T * self.timestep_scale
    x_B_C_T_H_W = _maybe_concat_condition_mask(
        self,
        x_B_C_T_H_W,
        is_video=data_type == DataType.VIDEO,
        condition_video_input_mask_B_C_T_H_W=condition_video_input_mask_B_C_T_H_W,
    )

    x_B_T_H_W_D, rope_emb_L_1_1_D, extra_pos_emb = self.prepare_embedded_sequence(
        x_B_C_T_H_W,
        fps=fps,
        padding_mask=padding_mask,
    )

    if self.crossattn_proj is not None:
        crossattn_emb = self.crossattn_proj(crossattn_emb)

    if img_context_emb is not None:
        assert self.extra_image_context_dim is not None
        img_context_emb = self.img_context_proj(img_context_emb)
        context_input = (crossattn_emb, img_context_emb)
    else:
        context_input = crossattn_emb

    with amp.autocast("cuda", enabled=getattr(self, "use_wan_fp32_strategy", False), dtype=torch.float32):
        if timesteps_B_T.ndim == 1:
            timesteps_B_T = timesteps_B_T.unsqueeze(1)
        t_embedding_B_T_D, adaln_lora_B_T_3D = self.t_embedder(timesteps_B_T)
        t_embedding_B_T_D, adaln_lora_B_T_3D = _maybe_embed_action(
            self,
            t_embedding_B_T_D,
            adaln_lora_B_T_3D,
            kwargs,
        )
        t_embedding_B_T_D = self.t_embedding_norm(t_embedding_B_T_D)

    self.affline_scale_log_info = {"t_embedding_B_T_D": t_embedding_B_T_D.detach()}
    self.affline_emb = t_embedding_B_T_D
    self.crossattn_emb = crossattn_emb

    if extra_pos_emb is not None:
        assert x_B_T_H_W_D.shape == extra_pos_emb.shape

    block_kwargs = {
        "emb_B_T_D": t_embedding_B_T_D,
        "crossattn_emb": context_input,
        "rope_emb_L_1_1_D": rope_emb_L_1_1_D,
        "adaln_lora_B_T_3D": adaln_lora_B_T_3D,
        "extra_per_block_pos_emb": extra_pos_emb,
    }

    skip_forward = False
    ori_x = x_B_T_H_W_D
    residual_x = None
    current_idx = self.cnt % 2
    test_x = x_B_T_H_W_D.clone()

    if self.cnt >= int(self.dicache_num_steps * self.dicache_ret_ratio):
        for blk in self.blocks[: self.dicache_probe_depth]:
            test_x = blk(test_x, **block_kwargs)

        if self.previous_input[current_idx] is not None and self.previous_internal_states[current_idx] is not None:
            delta_y = (test_x - self.previous_internal_states[current_idx]).abs().mean() / (
                self.previous_internal_states[current_idx].abs().mean() + 1e-8
            )
            self.accumulated_rel_l1_distance[current_idx] += delta_y

            if self.accumulated_rel_l1_distance[current_idx] < self.dicache_rel_l1_thresh:
                skip_forward = True
                self.resume_flag[current_idx] = False
                residual_x = self.residual_cache[current_idx]
            else:
                self.resume_flag[current_idx] = True
                self.accumulated_rel_l1_distance[current_idx] = 0
                self.previous_internal_states[current_idx] = test_x.clone()

    if skip_forward:
        if len(self.residual_window[current_idx]) >= 2:
            current_residual = test_x - x_B_T_H_W_D
            numer = (current_residual - self.probe_residual_window[current_idx][-2]).abs().mean()
            denom = (
                self.probe_residual_window[current_idx][-1]
                - self.probe_residual_window[current_idx][-2]
            ).abs().mean()
            gamma = (numer / denom).clip(1, 2) if denom > 1e-6 else 1.0
            x_B_T_H_W_D = x_B_T_H_W_D + self.residual_window[current_idx][-2] + gamma * (
                self.residual_window[current_idx][-1] - self.residual_window[current_idx][-2]
            )
        else:
            x_B_T_H_W_D = x_B_T_H_W_D + residual_x

        self.previous_internal_states[current_idx] = test_x
        self.previous_input[current_idx] = ori_x
    else:
        if self.resume_flag[current_idx]:
            x_B_T_H_W_D = test_x
            remaining = self.blocks[self.dicache_probe_depth :]
        else:
            remaining = self.blocks

        for i, blk in enumerate(remaining):
            x_B_T_H_W_D = blk(x_B_T_H_W_D, **block_kwargs)
            real_idx = i if not self.resume_flag[current_idx] else i + self.dicache_probe_depth
            if real_idx == self.dicache_probe_depth - 1:
                self.previous_internal_states[current_idx] = x_B_T_H_W_D.clone()

        residual_x = x_B_T_H_W_D - ori_x
        self.residual_cache[current_idx] = residual_x
        probe_residual = (
            self.previous_internal_states[current_idx] - ori_x
            if self.previous_internal_states[current_idx] is not None
            else residual_x
        )
        self.previous_input[current_idx] = ori_x

        if len(self.residual_window[current_idx]) <= 2:
            self.residual_window[current_idx].append(residual_x)
            self.probe_residual_window[current_idx].append(probe_residual)
        else:
            self.residual_window[current_idx][-2] = self.residual_window[current_idx][-1]
            self.residual_window[current_idx][-1] = residual_x
            self.probe_residual_window[current_idx][-2] = self.probe_residual_window[current_idx][-1]
            self.probe_residual_window[current_idx][-1] = probe_residual

    x_out = self.final_layer(
        x_B_T_H_W_D,
        t_embedding_B_T_D,
        adaln_lora_B_T_3D=adaln_lora_B_T_3D,
    )
    x_B_C_Tt_Hp_Wp = self.unpatchify(x_out)

    self.cnt += 1
    if self.cnt >= self.dicache_num_steps:
        initialize_dicache_state(self, num_steps=self.dicache_num_steps)

    return x_B_C_Tt_Hp_Wp


def apply_dicache(model, config: DiCacheConfig):
    model.dicache_enabled = True
    model.dicache_num_steps = config.num_steps
    model.dicache_rel_l1_thresh = config.rel_l1_thresh
    model.dicache_ret_ratio = config.ret_ratio
    model.dicache_probe_depth = config.probe_depth

    initialize_dicache_state(model, num_steps=config.num_steps)
    prepare_cache_runtime_model(model)
    model.forward = types.MethodType(dicache_mini_train_dit_forward, model)

    log.info(
        f"[DiCache] Applied: steps={config.num_steps} thresh={config.rel_l1_thresh} "
        f"ret_ratio={config.ret_ratio} probe_depth={config.probe_depth}"
    )
    return model