"""
"""

from typing import Any
from typing import Callable
from typing import ParamSpec

import spaces
import torch
from torch.utils._pytree import tree_map_only
from torchao.quantization import quantize_
from torchao.quantization import Float8DynamicActivationFloat8WeightConfig
from torchao.quantization import Int8WeightOnlyConfig

from optimization_utils import capture_component_call
from optimization_utils import aoti_compile
from optimization_utils import drain_module_parameters


P = ParamSpec('P')

# --- 新的、更精确的动态塑形定义 ---

# VAE 时间缩放因子为 1，latent_frames = num_frames。范围是 [8, 81]。
LATENT_FRAMES_DIM = torch.export.Dim('num_latent_frames', min=8, max=81)

# Transformer 的 patch_size 为 (1, 2, 2)，这意味着输入潜像的高度和宽度
# 实际上被除以 2。如果符号追踪器假设奇数是可能的，这会产生约束失败。
#
# 为了解决这个问题，我们为 *打过补丁后* (即除法后) 的尺寸定义动态维度，
# 然后将输入形状表示为该维度的 2 倍。这在数学上向编译器保证了
# 输入潜像维度始终为偶数，从而满足约束。

# 应用的像素尺寸范围：[480, 832]。VAE 缩放因子为 8。
# 潜像维度范围：[480/8, 832/8] = [60, 104]。
# 打过补丁后的潜像维度范围：[60/2, 104/2] = [30, 52]。
LATENT_PATCHED_HEIGHT_DIM = torch.export.Dim('latent_patched_height', min=30, max=52)
LATENT_PATCHED_WIDTH_DIM = torch.export.Dim('latent_patched_width', min=30, max=52)

# 现在，我们为 transformer 的 `hidden_states` 输入定义动态形状，
# 其形状为 (batch_size, channels, num_frames, height, width)。
TRANSFORMER_DYNAMIC_SHAPES = {
    'hidden_states': {
        2: LATENT_FRAMES_DIM,
        3: 2 * LATENT_PATCHED_HEIGHT_DIM, # 保证高度为偶数
        4: 2 * LATENT_PATCHED_WIDTH_DIM,  # 保证宽度为偶数
    },
}

# --- 定义结束 ---


INDUCTOR_CONFIGS = {
    'conv_1x1_as_mm': True,
    'epilogue_fusion': False,
    'coordinate_descent_tuning': True,
    'coordinate_descent_check_all_directions': True,
    'max_autotune': True,
    'triton.cudagraphs': True,
}


def optimize_pipeline_(pipeline: Callable[P, Any], *args: P.args, **kwargs: P.kwargs):

    @spaces.GPU(duration=1500)
    def compile_transformer():
        
        # LoRA 融合部分保持不变
        pipeline.load_lora_weights(
            "Kijai/WanVideo_comfy", 
            weight_name="Lightx2v/lightx2v_I2V_14B_480p_cfg_step_distill_rank128_bf16.safetensors", 
            adapter_name="lightx2v"
        )
        kwargs_lora = {}
        kwargs_lora["load_into_transformer_2"] = True
        pipeline.load_lora_weights(
            "Kijai/WanVideo_comfy", 
            weight_name="Lightx2v/lightx2v_I2V_14B_480p_cfg_step_distill_rank128_bf16.safetensors", 
            adapter_name="lightx2v_2", **kwargs_lora
        )
        pipeline.set_adapters(["lightx2v", "lightx2v_2"], adapter_weights=[1., 1.])
        pipeline.fuse_lora(adapter_names=["lightx2v"], lora_scale=3., components=["transformer"])
        pipeline.fuse_lora(adapter_names=["lightx2v_2"], lora_scale=1., components=["transformer_2"])
        pipeline.unload_lora_weights()
        
        # 捕获单次调用以获取 args/kwargs 结构
        with capture_component_call(pipeline, 'transformer') as call:
            pipeline(*args, **kwargs)
        
        dynamic_shapes = tree_map_only((torch.Tensor, bool), lambda t: None, call.kwargs)
        dynamic_shapes |= TRANSFORMER_DYNAMIC_SHAPES

        # 量化保持不变
        quantize_(pipeline.transformer, Float8DynamicActivationFloat8WeightConfig())
        quantize_(pipeline.transformer_2, Float8DynamicActivationFloat8WeightConfig())
        
        # --- 简化的编译流程 ---
        
        exported_1 = torch.export.export(
            mod=pipeline.transformer,
            args=call.args,
            kwargs=call.kwargs,
            dynamic_shapes=dynamic_shapes,
        )
        
        exported_2 = torch.export.export(
            mod=pipeline.transformer_2,
            args=call.args,
            kwargs=call.kwargs,
            dynamic_shapes=dynamic_shapes,
        )

        compiled_1 = aoti_compile(exported_1, INDUCTOR_CONFIGS)
        compiled_2 = aoti_compile(exported_2, INDUCTOR_CONFIGS)
        
        # 返回两个已编译的模型
        return compiled_1, compiled_2


    # 量化文本编码器 (与之前相同)
    quantize_(pipeline.text_encoder, Int8WeightOnlyConfig())
    
    # 获取两个经过动态塑形的已编译模型
    compiled_transformer_1, compiled_transformer_2 = compile_transformer()

    # --- 简化的赋值流程 ---
    
    pipeline.transformer.forward = compiled_transformer_1
    drain_module_parameters(pipeline.transformer)

    pipeline.transformer_2.forward = compiled_transformer_2
    drain_module_parameters(pipeline.transformer_2)