#!/usr/bin/env python3
"""Test quantization on single shard only."""

import gc
import torch
from pathlib import Path
from safetensors.torch import load_file, save_file
from bitsandbytes.functional import quantize_nf4
from transformers import AutoConfig


def test_single_shard(model_path, output_dir):
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    config = AutoConfig.from_pretrained(model_path, trust_remote_code=True)

    import glob
    shards = sorted(glob.glob(f"{model_path}/*.safetensors"))
    
    if len(shards) == 0:
        print("No shards found!")
        return
    
    # Use only first shard
    shard_file = shards[0]
    print(f"Testing with single shard: {shard_file}")
    
    state_dict = load_file(shard_file, device="cpu")
    
    weight_keys = [
        k for k, v in state_dict.items()
        if "weight" in k and isinstance(v, torch.Tensor) and v.dim() == 2
    ]
    
    print(f"Found {len(weight_keys)} weight tensors\n")
    
    quantized = 0
    failed = 0
    
    for key in weight_keys[:5]:  # Test first 5 layers
        try:
            weight = state_dict[key].to("cuda:0")
            qweight, qstate = quantize_nf4(weight, blocksize=64, compress_statistics=True)
            state_dict[key] = qweight.cpu()
            quantized += 1
            print(f"✓ {key}")
        except Exception as e:
            failed += 1
            print(f"✗ {key}: {e}")
    
    print(f"\nResults: {quantized} quantized, {failed} failed")
    
    # Save test output
    test_output = output_dir / "test_shard.safetensors"
    state_dict_cpu = {k: v.cpu() if isinstance(v, torch.Tensor) else v for k, v in state_dict.items()}
    save_file(state_dict_cpu, test_output)
    print(f"Saved test output to: {test_output}")


if __name__ == "__main__":
    test_single_shard("/data/models/Ornith-1.0-35B", "/data/models/test_quantize")