# examples/swin_export.py
"""Export genuine native TorchVision Swin-T V1 classification without network.

Supply the checkpoint explicitly and licensed local RGB photos with provenance.
The native TorchVision backbone is used directly; no custom-backbone parity or
original-paper benchmark reproduction is claimed. All 1000 outputs and the exact
normalized inputs are preserved. Install only requirements-swin.txt if needed.
"""

import argparse
import hashlib
import inspect
import json
import platform
import shutil
import sys
from pathlib import Path

import numpy as np
import torch
import torchvision
import torchvision.models.swin_transformer as native_swin
from PIL import Image
from PIL import __version__ as pillow_version
from torchvision import transforms
from torchvision.models import Swin_T_Weights, swin_t

CHECKPOINT_URL = "https://download.pytorch.org/models/swin_t-704ceda3.pth"
CHECKPOINT_SHA256 = "704ceda373461b0a224fcdddd75cd2a5e9f8064512ed47adbddef7f343fd147b"
ORIGINAL_REVISION = "f82860bfb5225915aca09c3227159ee9e1df874d"


def digest(path):
    """Hash file bytes without buffering a large checkpoint."""
    with Path(path).open("rb") as stream:
        return hashlib.file_digest(stream, "sha256").hexdigest()


def export(checkpoint, images_dir, photo_sources, output_dir):
    """Run the native full classifier and preserve exact input/output evidence."""
    if digest(checkpoint) != CHECKPOINT_SHA256:
        raise ValueError("Swin-T V1 checkpoint checksum mismatch")
    torch.set_num_threads(2)
    torch.manual_seed(13)
    weights = Swin_T_Weights.IMAGENET1K_V1
    model = swin_t(weights=None).eval()
    model.load_state_dict(
        torch.load(checkpoint, map_location="cpu", weights_only=True), strict=True
    )
    preprocessing = weights.transforms()
    records = json.loads(Path(photo_sources).read_text())
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    source_module = Path(inspect.getfile(native_swin))
    presets_module = Path(inspect.getfile(type(preprocessing)))
    functional_module = Path(inspect.getfile(transforms.functional))
    pillow_module = Path(inspect.getfile(transforms.functional.F_pil))
    package = Path(torchvision.__file__).parent
    native_sources = {
        str(path.relative_to(package)): digest(path)
        for path in (source_module, presets_module, functional_module, pillow_module)
    }
    program = Path(__file__).with_name("swin_transformer_training.py")
    original_manifest = Path(__file__).with_name("swin-original-source-hashes.json")
    result = {
        "provenance": {
            "model": "torchvision.models.swin_t (Swin V1, not Swin V2)",
            "architecture": {
                "patchSize": 4,
                "windowSize": 7,
                "width": 96,
                "depths": [2, 2, 6, 2],
                "heads": [3, 6, 12, 24],
            },
            "weights": "Swin_T_Weights.IMAGENET1K_V1",
            "checkpointUrl": CHECKPOINT_URL,
            "checkpointSha256": digest(checkpoint),
            "parameterCount": sum(p.numel() for p in model.parameters()),
            "strictStateDict": True,
            "device": "cpu",
            "dtype": "float32",
            "python": platform.python_version(),
            "torch": torch.__version__,
            "torchvision": torchvision.__version__,
            "numpy": np.__version__,
            "pillow": pillow_version,
            "torchvisionRevision": torchvision.version.git_version,
            "torchvisionSourceHashes": native_sources,
            "originalMicrosoftRevision": ORIGINAL_REVISION,
            "originalSourceManifestSha256": digest(original_manifest),
            "programSha256": digest(program),
            "exportSha256": digest(__file__),
            "photoSourcesSha256": digest(photo_sources),
            "command": [sys.executable, *sys.argv],
            "trainingRecipe": weights.meta["recipe"],
            "publishedTorchvisionTop1": weights.meta["_metrics"]["ImageNet-1K"][
                "acc@1"
            ],
            "originalPaperSwinTTop1": 81.3,
            "scope": (
                "Two real-photo native inferences; no ImageNet evaluation "
                "or original benchmark reproduction"
            ),
        },
        "preprocessing": {
            "resizeSize": preprocessing.resize_size,
            "cropSize": preprocessing.crop_size,
            "interpolation": preprocessing.interpolation.value,
            "antialias": preprocessing.antialias,
            "mean": preprocessing.mean,
            "std": preprocessing.std,
            "rescale": "PIL RGB uint8 to float32 [0,1], then (x-mean)/std",
        },
        "categories": weights.meta["categories"],
        "photos": [],
    }
    for photo in records:
        source = Path(images_dir) / photo["originalImage"]
        if digest(source) != photo["source"]["sha256"]:
            raise ValueError(f"Photo checksum mismatch: {source}")
        image = Image.open(source).convert("RGB")
        tensor = preprocessing(image).unsqueeze(0)
        processed = transforms.functional.center_crop(
            transforms.functional.resize(
                image,
                preprocessing.resize_size,
                interpolation=preprocessing.interpolation,
                antialias=preprocessing.antialias,
            ),
            preprocessing.crop_size,
        )
        # Authenticate the visual crop against the exact normalized model input.
        explicit = transforms.functional.normalize(
            transforms.functional.pil_to_tensor(processed).float() / 255,
            preprocessing.mean,
            preprocessing.std,
        ).unsqueeze(0)
        if not torch.equal(tensor, explicit):
            raise AssertionError("Processed photo differs from native input")
        with torch.inference_mode():
            logits = model(tensor)[0]
            probabilities = logits.softmax(0)
        scores, indices = probabilities.topk(5)
        stem = photo["id"]
        processed_name, arrays_name = f"input-{stem}.png", f"arrays-{stem}.npz"
        processed.save(output_dir / processed_name)
        shutil.copyfile(source, output_dir / photo["originalImage"])
        np.savez_compressed(
            output_dir / arrays_name,
            normalized_input=tensor.numpy(),
            logits=logits.numpy(),
            probabilities=probabilities.numpy(),
        )
        result["photos"].append(
            {
                **photo,
                "originalShape": [image.height, image.width],
                "inputImage": processed_name,
                "inputShape": list(tensor.shape),
                "inputImageSha256": digest(output_dir / processed_name),
                "normalizedInputSha256": hashlib.sha256(
                    tensor.numpy().tobytes()
                ).hexdigest(),
                "processedInputParityMaxError": float((explicit - tensor).abs().max()),
                "arrays": arrays_name,
                "arraysSha256": digest(output_dir / arrays_name),
                "logits": logits.tolist(),
                "probabilities": probabilities.tolist(),
                "top5": [
                    {
                        "index": int(index),
                        "label": weights.meta["categories"][index],
                        "score": float(score),
                    }
                    for score, index in zip(scores, indices)
                ],
            }
        )
    (output_dir / "predictions.json").write_text(json.dumps(result, indent=2) + "\n")
    return result


def main():
    """Read explicit local inputs, never download model weights or photos."""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--checkpoint", required=True, type=Path)
    parser.add_argument("--images-dir", required=True, type=Path)
    parser.add_argument(
        "--photo-sources",
        type=Path,
        default=Path(__file__).with_name("swin-photo-sources.json"),
    )
    parser.add_argument("--output-dir", required=True, type=Path)
    args = parser.parse_args()
    result = export(
        args.checkpoint, args.images_dir, args.photo_sources, args.output_dir
    )
    print(
        json.dumps(
            {
                "photos": len(result["photos"]),
                "parameterCount": result["provenance"]["parameterCount"],
            }
        )
    )


if __name__ == "__main__":
    main()
