methods/cache_strategy/FasterCache/runtime.py

from __future__ import annotations

import math
import types
from dataclasses import dataclass, field
from typing import Any

import torch

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()

from methods.cache_strategy.common import FasterCacheConfig


def resolve_fastercache_start_step(config: FasterCacheConfig, num_steps: int) -> int:
    if config.start_step > 0:
        return config.start_step
    return int(math.ceil(max(num_steps, 1) * 0.3))


def _history_for_shape(history: list[torch.Tensor], shape: torch.Size) -> bool:
    return len(history) >= 2 and history[-1].shape == shape and history[-2].shape == shape


def _split_fft_bands(x: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor]:
    flat = x.reshape(x.shape[0], -1)
    if flat.dtype == torch.bfloat16:
        flat = flat.to(torch.float32)
    freq = torch.fft.rfft(flat, dim=-1)
    split_idx = max(freq.shape[-1] // 4, 1)
    lf = freq.clone()
    hf = freq.clone()
    lf[..., split_idx:] = 0
    hf[..., :split_idx] = 0
    return lf, hf


def _compose_from_fft_bands(
    reference: torch.Tensor,
    delta_lf: torch.Tensor,
    delta_hf: torch.Tensor,
) -> torch.Tensor:
    ref_flat = reference.reshape(reference.shape[0], -1)
    if ref_flat.dtype == torch.bfloat16:
        ref_flat = ref_flat.to(torch.float32)
    ref_freq = torch.fft.rfft(ref_flat, dim=-1)
    recon = torch.fft.irfft(ref_freq + delta_lf + delta_hf, n=ref_flat.shape[-1], dim=-1)
    return recon.reshape_as(reference).type_as(reference)


def _clone_detached(x: torch.Tensor) -> torch.Tensor:
    return x.detach().clone()


def _clone_detached_cpu(x: torch.Tensor) -> torch.Tensor:
    return x.detach().to("cpu", copy=True)


def _materialize_to_device(x: torch.Tensor, device: torch.device) -> torch.Tensor:
    if x.device == device:
        return x
    return x.to(device, non_blocking=False)


def compute_fastercache_deltas(
    conditional: torch.Tensor,
    unconditional: torch.Tensor,
) -> tuple[torch.Tensor, torch.Tensor]:
    lf_cond, hf_cond = _split_fft_bands(conditional)
    lf_uncond, hf_uncond = _split_fft_bands(unconditional)
    return _clone_detached(lf_uncond - lf_cond), _clone_detached(hf_uncond - hf_cond)


def reconstruct_fastercache_output(
    conditional: torch.Tensor,
    delta_lf: torch.Tensor,
    delta_hf: torch.Tensor,
) -> torch.Tensor:
    return _compose_from_fft_bands(conditional, delta_lf, delta_hf)


@dataclass
class FasterCacheRuntimeState:
    config: FasterCacheConfig
    cfg_mode: str
    total_steps: int = 0
    resolved_start_step: int = 0
    call_idx: int = 0
    current_step_idx: int = 0
    current_branch: str = "cond"
    pending_cond_output: torch.Tensor | None = None
    delta_lf: torch.Tensor | None = None
    delta_hf: torch.Tensor | None = None
    block_histories: dict[tuple[str, int], list[torch.Tensor]] = field(default_factory=dict)
    block_seq_len_min: int = 2048
    block_alpha: float = 0.3
    skipped_timesteps_count: int = 0

    def reset(self, num_steps: int) -> None:
        self.total_steps = int(num_steps)
        self.resolved_start_step = resolve_fastercache_start_step(self.config, num_steps)
        self.call_idx = 0
        self.current_step_idx = 0
        self.current_branch = "cond"
        self.pending_cond_output = None
        self.delta_lf = None
        self.delta_hf = None
        self.block_histories.clear()
        self.skipped_timesteps_count = 0

    def is_model_anchor_step(self, step_idx: int) -> bool:
        if step_idx < self.resolved_start_step:
            return True
        interval = max(int(self.config.model_interval), 1)
        return (step_idx - self.resolved_start_step) % interval == 0

    def is_block_anchor_step(self, step_idx: int) -> bool:
        if step_idx < self.resolved_start_step:
            return True
        interval = max(int(self.config.block_interval), 1)
        return (step_idx - self.resolved_start_step) % interval == 0


def initialize_fastercache_state(model, config: FasterCacheConfig, *, cfg_mode: str) -> FasterCacheRuntimeState:
    state = FasterCacheRuntimeState(config=config, cfg_mode=cfg_mode)
    state.reset(num_steps=1)
    model.fastercache_state = state
    model.fastercache_enabled = True
    model.fastercache_config = config
    model.has_fastercache_backend = True
    return state


def reset_fastercache_state(model, config: FasterCacheConfig, num_steps: int) -> int:
    state = getattr(model, "fastercache_state", None)
    if state is None:
        state = initialize_fastercache_state(model, config, cfg_mode="sequential")
    state.reset(num_steps=num_steps)
    return num_steps


def _get_dit_block_start_idx(num_blocks: int) -> int:
    if num_blocks >= 36:
        return 6
    if num_blocks >= 28:
        return 4
    if num_blocks >= 20:
        return 3
    return max(2, num_blocks // 6)


def _dit_layer_is_eligible(model, layer_idx: int) -> bool:
    num_blocks = int(getattr(model, "num_blocks", len(getattr(model, "blocks", []))))
    start_idx = _get_dit_block_start_idx(num_blocks)
    return start_idx <= layer_idx < max(num_blocks - 2, start_idx)


def _run_dit_model_shortcut(state: FasterCacheRuntimeState) -> bool:
    return (
        state.current_branch == "uncond"
        and not state.is_model_anchor_step(state.current_step_idx)
        and state.pending_cond_output is not None
        and state.delta_lf is not None
        and state.delta_hf is not None
    )


def _patch_dit_block_self_attention(model) -> None:
    state: FasterCacheRuntimeState = model.fastercache_state
    for layer_idx, block in enumerate(model.blocks):
        attn = getattr(block, "self_attn", None)
        if attn is None or hasattr(attn, "_fastercache_original_forward"):
            continue

        original_forward = attn.forward
        attn._fastercache_original_forward = original_forward
        attn._fastercache_layer_idx = layer_idx

        def _wrapped_forward(
            self,
            x: torch.Tensor,
            context: torch.Tensor | None = None,
            rope_emb: torch.Tensor | None = None,
            video_size=None,
            attention_runtime_context: dict | None = None,
            **kwargs,
        ):
            del attention_runtime_context
            if context is not None:
                return self._fastercache_original_forward(
                    x,
                    context=context,
                    rope_emb=rope_emb,
                    video_size=video_size,
                    **kwargs,
                )

            runtime_state: FasterCacheRuntimeState = model.fastercache_state
            seq_len = int(x.shape[1])
            layer_history = runtime_state.block_histories.setdefault(
                (runtime_state.current_branch, self._fastercache_layer_idx), []
            )

            can_reuse = (
                _dit_layer_is_eligible(model, self._fastercache_layer_idx)
                and seq_len >= runtime_state.block_seq_len_min
                and not runtime_state.is_block_anchor_step(runtime_state.current_step_idx)
                and _history_for_shape(layer_history, x.shape)
            )
            if can_reuse:
                latest = _materialize_to_device(layer_history[-1], x.device).type_as(x)
                previous = _materialize_to_device(layer_history[-2], x.device).type_as(x)
                return latest + (latest - previous) * runtime_state.block_alpha

            out = self._fastercache_original_forward(
                x,
                context=context,
                rope_emb=rope_emb,
                video_size=video_size,
                **kwargs,
            )
            layer_history.append(_clone_detached_cpu(out))
            if len(layer_history) > 2:
                del layer_history[:-2]
            return out

        attn.forward = types.MethodType(_wrapped_forward, attn)


def apply_fastercache(model, config: FasterCacheConfig, *, cfg_mode: str = "sequential"):
    if hasattr(model, "_fastercache_original_forward"):
        initialize_fastercache_state(model, config, cfg_mode=cfg_mode)
        return model

    state = initialize_fastercache_state(model, config, cfg_mode=cfg_mode)
    original_forward = model.forward
    model._fastercache_original_forward = original_forward
    _patch_dit_block_self_attention(model)

    def _wrapped_forward(self, *args, **kwargs):
        runtime_state: FasterCacheRuntimeState = self.fastercache_state
        if kwargs.get("use_cuda_graphs", False):
            raise ValueError("[FasterCache] FasterCache is incompatible with CUDA Graphs in v1.")

        runtime_state.current_step_idx = runtime_state.call_idx // 2
        runtime_state.current_branch = "cond" if runtime_state.call_idx % 2 == 0 else "uncond"
        model_shortcut_eligible = not runtime_state.is_model_anchor_step(runtime_state.current_step_idx)

        if _run_dit_model_shortcut(runtime_state):
            if runtime_state.config.debug:
                log.info(f"[FasterCache] Step {runtime_state.current_step_idx} ({runtime_state.current_branch}): Model forward SKIPPED completely")
            runtime_state.skipped_timesteps_count += 1
            
            # Move needed tensors back to GPU for reconstruction
            device = next(self.parameters()).device
            p_cond = _materialize_to_device(runtime_state.pending_cond_output, device)
            d_lf = _materialize_to_device(runtime_state.delta_lf, device)
            d_hf = _materialize_to_device(runtime_state.delta_hf, device)
            
            output = _compose_from_fft_bands(p_cond, d_lf, d_hf)
        else:
            if runtime_state.config.debug:
                if runtime_state.is_block_anchor_step(runtime_state.current_step_idx):
                    log.info(f"[FasterCache] Step {runtime_state.current_step_idx} ({runtime_state.current_branch}): Model EXECUTED (Anchor Step - Blocks COMPUTED)")
                else:
                    log.info(f"[FasterCache] Step {runtime_state.current_step_idx} ({runtime_state.current_branch}): Model EXECUTED (Blocks REUSED)")
            output = self._fastercache_original_forward(*args, **kwargs)

        if runtime_state.current_branch == "cond":
            if model_shortcut_eligible:
                # Only keep the conditional output when the upcoming unconditional pass may use it.
                runtime_state.pending_cond_output = _clone_detached_cpu(output)
                torch.cuda.empty_cache()
            else:
                runtime_state.pending_cond_output = None
                runtime_state.delta_lf = None
                runtime_state.delta_hf = None
        elif model_shortcut_eligible and runtime_state.pending_cond_output is not None:
            # Compute/update FFT deltas only on steps where future model shortcut may be used.
            device = output.device
            cond_output = _materialize_to_device(runtime_state.pending_cond_output, device).type_as(output)
            dlf, dhf = compute_fastercache_deltas(cond_output, output)
            
            # Store results on CPU
            runtime_state.delta_lf = dlf.to("cpu")
            runtime_state.delta_hf = dhf.to("cpu")
            torch.cuda.empty_cache()
            
            runtime_state.pending_cond_output = None
            if runtime_state.config.debug:
                log.info(f"[FasterCache] Step {runtime_state.current_step_idx}: Computed FFT deltas (all cache on CPU)")
                
            if runtime_state.current_step_idx >= runtime_state.total_steps - 1:
                log.info(f"[FasterCache] Generation completed. Total timesteps completely skipped: {runtime_state.skipped_timesteps_count}")
        elif runtime_state.current_branch == "uncond":
            runtime_state.pending_cond_output = None

        runtime_state.call_idx += 1
        return output

    model.forward = types.MethodType(_wrapped_forward, model)
    log.info(
        f"[FasterCache] Applied DiT runtime: start_step={config.start_step} model_interval={config.model_interval} block_interval={config.block_interval} cfg_mode={cfg_mode}"
    )
    return model