# examples/faster_rcnn_export.py
"""Export real COCO_V1 native detector outputs from local licensed photos.

No checkpoint download, training, or benchmark evaluation is performed.
Run beside faster_rcnn_training.py with the pinned requirements.
"""

import argparse
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
import platform
import sys

from PIL import Image, __version__ as PILLOW_VERSION
import torch
import torchvision
from torchvision.models.detection import (
    FasterRCNN,
    FasterRCNN_ResNet50_FPN_Weights,
)
from torchvision.models.detection.backbone_utils import resnet_fpn_backbone
from torchvision.models.detection._utils import overwrite_eps
from torchvision.ops import FrozenBatchNorm2d

CHECKPOINT_SHA = (
    "258fb6c638b15964ddcdd1ae0748c5eef1be9e732750120cc857feed3faac384"
)
REVISION = "78839c2b06c83c6cfb5c4da692ffb331bbd4c4cc"


def digest(path):
    """Hash exact local file bytes."""
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


def load_detector(checkpoint):
    """Rebuild the native COCO_V1 architecture and strictly load all state keys."""
    if digest(checkpoint) != CHECKPOINT_SHA:
        raise ValueError(
            "Checkpoint SHA256 does not match the COCO_V1 release"
        )
    # Exactly the native pretrained builder's backbone normalization. Building
    # with weights=None alone selects BatchNorm instead of FrozenBatchNorm.
    backbone = resnet_fpn_backbone(
        backbone_name="resnet50",
        weights=None,
        trainable_layers=3,
        norm_layer=FrozenBatchNorm2d,
    )
    model = FasterRCNN(backbone, num_classes=91)
    state = torch.load(checkpoint, map_location="cpu", weights_only=True)
    loaded = model.load_state_dict(state, strict=True)
    overwrite_eps(
        model, 0.0
    )  # Native COCO_V1 builder's exact FrozenBN epsilon.
    return model.cpu().float().eval(), {
        "strict": True,
        "state_keys": len(state),
        "missing_keys": loaded.missing_keys,
        "unexpected_keys": loaded.unexpected_keys,
    }


def export(checkpoint, photos_dir, output):
    """Run the full CPU detector and retain every final box, score, and label."""
    for actual, expected in [
        (torch.__version__, "2.12.0"),
        (torchvision.__version__, "0.27.0"),
        (PILLOW_VERSION, "12.2.0"),
    ]:
        if actual.split("+")[0] != expected:
            raise ValueError(f"Expected {expected}, got {actual}")
    if torchvision.version.git_version != REVISION:
        raise ValueError(
            "TorchVision source revision does not match pinned native implementation"
        )
    torch.set_num_threads(2)
    torch.manual_seed(8)
    weights = FasterRCNN_ResNet50_FPN_Weights.COCO_V1
    model, loading = load_detector(checkpoint)
    photo_manifest = json.loads((photos_dir / "predictions.json").read_text())
    output.mkdir(parents=True, exist_ok=True)
    records = []
    for photo in photo_manifest["photos"]:
        image_path = photos_dir / photo["originalImage"]
        if digest(image_path) != photo["source"]["sha256"]:
            raise ValueError(
                "Photo bytes differ from licensed source manifest"
            )
        with Image.open(image_path) as source:
            image = source.convert("RGB")
        tensor = weights.transforms()(image)
        # Read the real evaluation transform output shape, then let the complete
        # detector independently execute that same normalization/resize/padding.
        with torch.inference_mode():
            transformed, _ = model.transform([tensor], None)
            prediction = model([tensor])[0]
        boxes, labels, scores = (
            prediction["boxes"],
            prediction["labels"],
            prediction["scores"],
        )
        if (
            not torch.isfinite(boxes).all()
            or not torch.isfinite(scores).all()
            or not ((boxes[:, 2:] > boxes[:, :2]).all())
        ):
            raise RuntimeError("Native inference produced invalid outputs")
        records.append(
            {
                "id": photo["id"],
                "title": photo["title"],
                "image": f"/examples/vit-demo/{photo['originalImage']}",
                "original_shape": [image.height, image.width],
                "source": photo["source"],
                "input_tensor_shape": list(tensor.shape),
                "input_tensor_sha256": hashlib.sha256(
                    tensor.numpy().tobytes()
                ).hexdigest(),
                "resized_shape": list(transformed.image_sizes[0]),
                "padded_batch_shape": list(transformed.tensors.shape),
                "detections": [
                    {
                        "box": box.tolist(),
                        "label_index": int(label),
                        "label": weights.meta["categories"][int(label)],
                        "score": float(score),
                    }
                    for box, label, score in zip(boxes, labels, scores)
                ],
            }
        )
    source_files = []
    root = Path(torchvision.__file__).parent
    for relative in [
        "models/detection/faster_rcnn.py",
        "models/detection/backbone_utils.py",
        "models/detection/rpn.py",
        "models/detection/roi_heads.py",
        "models/detection/transform.py",
        "ops/poolers.py",
    ]:
        source_files.append(
            {
                "path": relative,
                "sha256": digest(root / relative),
                "url": f"https://github.com/pytorch/vision/blob/{REVISION}/torchvision/{relative}",
            }
        )
    result = {
        "provenance": {
            "scope": "Native later TorchVision ResNet-50-FPN Faster R-CNN inference; not original VGG16 benchmark reproduction",
            "architecture": "FasterRCNN / ResNet-50-FPN / MultiScaleRoIAlign / 91 COCO category slots",
            "weights": "FasterRCNN_ResNet50_FPN_Weights.COCO_V1",
            "checkpoint_url": weights.url,
            "checkpoint_sha256": CHECKPOINT_SHA,
            "loading": loading,
            "parameter_count": sum(p.numel() for p in model.parameters()),
            "source_revision": REVISION,
            "source_files": source_files,
            "program_sha256": digest(Path(__file__)),
            "reference_sha256": digest(
                Path(__file__).with_name("faster_rcnn_training.py")
            ),
            "command": "python examples/faster_rcnn_export.py --checkpoint .context/paper-series/weights/fasterrcnn_resnet50_fpn_coco-258fb6c6.pth --photos-dir public/examples/vit-demo --output public/examples/faster-rcnn-demo",
            "executed_argv": sys.argv,
            "generated_at": datetime.now(timezone.utc).isoformat(),
            "environment": {
                "python": platform.python_version(),
                "torch": torch.__version__,
                "torchvision": torchvision.__version__,
                "pillow": PILLOW_VERSION,
                "device": "cpu",
                "dtype": "float32",
                "threads": 2,
                "seed": 8,
            },
        },
        "preprocessing": {
            "input": "PIL RGB uint8 -> float32 [C,H,W] /255; no crop",
            "internal_mean": model.transform.image_mean,
            "internal_std": model.transform.image_std,
            "min_size": list(model.transform.min_size),
            "max_size": model.transform.max_size,
            "resize": "Tensor bilinear interpolation, align_corners=False, recompute_scale_factor=True; no antialias flag",
            "size_divisible": model.transform.size_divisible,
            "padding": "zeros after normalization at bottom/right to a multiple of 32",
            "boxes": "continuous xyxy coordinates rescaled to the original uncropped photograph",
        },
        "postprocessing": {
            "score_threshold": model.roi_heads.score_thresh,
            "classwise_nms_iou": model.roi_heads.nms_thresh,
            "max_detections": model.roi_heads.detections_per_img,
            "rpn_nms_iou": model.rpn.nms_thresh,
            "rpn_pre_nms_top_n": model.rpn.pre_nms_top_n(),
            "rpn_post_nms_top_n": model.rpn.post_nms_top_n(),
        },
        "photos": records,
    }
    temporary = output / "predictions.json.tmp"
    temporary.write_text(json.dumps(result, indent=2, allow_nan=False) + "\n")
    temporary.replace(output / "predictions.json")
    print(
        json.dumps(
            {
                "strict_loading": loading,
                "photos": [
                    {"id": photo["id"], "detections": len(photo["detections"])}
                    for photo in records
                ],
            }
        )
    )


def main():
    """Require explicit local inputs; never fetch models or images."""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--checkpoint", type=Path, required=True)
    parser.add_argument("--photos-dir", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    export(args.checkpoint, args.photos_dir, args.output)


if __name__ == "__main__":
    main()
