src/cond/cast.py

"""Weight casting utilities for efficient model loading."""
from src.Device import Device
import torch
import logging

logger = logging.getLogger(__name__)

def cast_to(weight, dtype=None, device=None, non_blocking=False, copy=False):
    """Cast a weight tensor to specified dtype and device."""
    if device is None or weight.device == device:
        if not copy and (dtype is None or weight.dtype == dtype):
            return weight
        return weight.to(dtype=dtype, copy=copy)
    r = torch.empty_like(weight, dtype=dtype, device=device)
    r.copy_(weight, non_blocking=non_blocking)
    return r


def cast_to_input(weight, input, non_blocking=False, copy=True):
    """Cast weight tensor to match input tensor's dtype and device."""
    return cast_to(weight, input.dtype, input.device, non_blocking=non_blocking, copy=copy)


def cast_bias_weight(s, input=None, dtype=None, device=None, bias_dtype=None):
    """Cast module's bias and weight to match input tensor."""
    if input is not None:
        dtype = dtype or input.dtype
        bias_dtype = bias_dtype or dtype
        device = device or input.device

    non_blocking = Device.device_supports_non_blocking(device)
    bias = None
    if s.bias is not None:
        has_fn = s.bias_function is not None
        bias = cast_to(s.bias, bias_dtype, device, non_blocking=non_blocking, copy=has_fn)
        if has_fn:
            bias = s.bias_function(bias)

    has_fn = s.weight_function is not None
    weight = cast_to(s.weight, None, device, non_blocking=non_blocking, copy=has_fn)
    
    # Handle NVFP4 dequantization
    if getattr(s, "quant_format", None) == "nvfp4":
        from src.Utilities.Quantization import dequantize_nvfp4
        weight = dequantize_nvfp4(
            weight, 
            s.weight_scale_2, 
            s.weight_scale, 
            s.original_shape
        )
        weight = weight.to(dtype)
    else:
        weight = weight.to(dtype)

    if has_fn:
        weight = s.weight_function(weight)
    return weight, bias


class CastWeightBiasOp:
    """Mixin for cast weight/bias operations."""
    comfy_cast_weights = False
    weight_function = None
    bias_function = None


class disable_weight_init:
    """Module wrappers with disabled weight initialization."""

    class Linear(torch.nn.Linear, CastWeightBiasOp):
        def reset_parameters(self): return None
        def forward_comfy_cast_weights(self, input):
            weight, bias = cast_bias_weight(self, input)
            return torch.nn.functional.linear(input, weight, bias)
        def forward(self, *args, **kwargs):
            return self.forward_comfy_cast_weights(*args, **kwargs) if self.comfy_cast_weights else super().forward(*args, **kwargs)

    class Conv1d(torch.nn.Conv1d, CastWeightBiasOp):
        def reset_parameters(self): return None
        def forward_comfy_cast_weights(self, input):
            weight, bias = cast_bias_weight(self, input)
            return self._conv_forward(input, weight, bias)
        def forward(self, *args, **kwargs):
            return self.forward_comfy_cast_weights(*args, **kwargs) if self.comfy_cast_weights else super().forward(*args, **kwargs)

    class Conv2d(torch.nn.Conv2d, CastWeightBiasOp):
        def reset_parameters(self): return None
        def forward_cast_weights(self, input):
            weight, bias = cast_bias_weight(self, input)
            return self._conv_forward(input, weight, bias)
        def forward(self, *args, **kwargs):
            return self.forward_cast_weights(*args, **kwargs) if self.comfy_cast_weights else super().forward(*args, **kwargs)

    class Conv3d(torch.nn.Conv3d, CastWeightBiasOp):
        def reset_parameters(self): return None
        def forward_comfy_cast_weights(self, input):
            weight, bias = cast_bias_weight(self, input)
            return self._conv_forward(input, weight, bias)
        def forward(self, *args, **kwargs):
            return self.forward_comfy_cast_weights(*args, **kwargs) if self.comfy_cast_weights else super().forward(*args, **kwargs)

    class GroupNorm(torch.nn.GroupNorm, CastWeightBiasOp):
        def reset_parameters(self): return None
        def forward_comfy_cast_weights(self, input):
            weight, bias = cast_bias_weight(self, input)
            return torch.nn.functional.group_norm(input, self.num_groups, weight, bias, self.eps)
        def forward(self, *args, **kwargs):
            return self.forward_comfy_cast_weights(*args, **kwargs) if self.comfy_cast_weights else super().forward(*args, **kwargs)

    class LayerNorm(torch.nn.LayerNorm, CastWeightBiasOp):
        def reset_parameters(self): return None
        def forward_cast_weights(self, input):
            weight, bias = cast_bias_weight(self, input) if self.weight is not None else (None, None)
            return torch.nn.functional.layer_norm(input, self.normalized_shape, weight, bias, self.eps)
        def forward(self, *args, **kwargs):
            return self.forward_cast_weights(*args, **kwargs) if self.comfy_cast_weights else super().forward(*args, **kwargs)

    class ConvTranspose2d(torch.nn.ConvTranspose2d, CastWeightBiasOp):
        def reset_parameters(self): return None
        def forward_comfy_cast_weights(self, input, output_size=None):
            output_padding = self._output_padding(input, output_size, self.stride, self.padding, self.kernel_size, 2, self.dilation)
            weight, bias = cast_bias_weight(self, input)
            return torch.nn.functional.conv_transpose2d(input, weight, bias, self.stride, self.padding, output_padding, self.groups, self.dilation)
        def forward(self, *args, **kwargs):
            return self.forward_comfy_cast_weights(*args, **kwargs) if self.comfy_cast_weights else super().forward(*args, **kwargs)

    class ConvTranspose1d(torch.nn.ConvTranspose1d, CastWeightBiasOp):
        def reset_parameters(self): return None
        def forward_comfy_cast_weights(self, input, output_size=None):
            output_padding = self._output_padding(input, output_size, self.stride, self.padding, self.kernel_size, 1, self.dilation)
            weight, bias = cast_bias_weight(self, input)
            return torch.nn.functional.conv_transpose1d(input, weight, bias, self.stride, self.padding, output_padding, self.groups, self.dilation)
        def forward(self, *args, **kwargs):
            return self.forward_comfy_cast_weights(*args, **kwargs) if self.comfy_cast_weights else super().forward(*args, **kwargs)

    class Embedding(torch.nn.Embedding, CastWeightBiasOp):
        def reset_parameters(self):
            self.bias = None
            return None
        def forward_comfy_cast_weights(self, input, out_dtype=None):
            output_dtype = out_dtype
            if self.weight.dtype in (torch.float16, torch.bfloat16):
                out_dtype = None
            weight, _ = cast_bias_weight(self, device=input.device, dtype=out_dtype)
            return torch.nn.functional.embedding(input, weight, self.padding_idx, self.max_norm, self.norm_type, self.scale_grad_by_freq, self.sparse).to(dtype=output_dtype)
        def forward(self, *args, **kwargs):
            if self.comfy_cast_weights:
                return self.forward_comfy_cast_weights(*args, **kwargs)
            kwargs.pop("out_dtype", None)
            return super().forward(*args, **kwargs)

    @classmethod
    def conv_nd(cls, dims, *args, **kwargs):
        """Create Conv2d/Conv3d based on dimensions."""
        if dims == 2: return cls.Conv2d(*args, **kwargs)
        if dims == 3: return cls.Conv3d(*args, **kwargs)
        raise ValueError(f"unsupported dimensions: {dims}")


class manual_cast(disable_weight_init):
    """Module wrappers with manual casting enabled by default."""
    class Linear(disable_weight_init.Linear): comfy_cast_weights = True
    class Conv1d(disable_weight_init.Conv1d): comfy_cast_weights = True
    class Conv2d(disable_weight_init.Conv2d): comfy_cast_weights = True
    class Conv3d(disable_weight_init.Conv3d): comfy_cast_weights = True
    class GroupNorm(disable_weight_init.GroupNorm): comfy_cast_weights = True
    class LayerNorm(disable_weight_init.LayerNorm): comfy_cast_weights = True
    class ConvTranspose2d(disable_weight_init.ConvTranspose2d): comfy_cast_weights = True
    class ConvTranspose1d(disable_weight_init.ConvTranspose1d): comfy_cast_weights = True
    class Embedding(disable_weight_init.Embedding): comfy_cast_weights = True