{"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"name":"python","version":"3.11"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# Selective DR grading: parallel preparation and reusable image cache\nAttach original EyePACS train/test archives and APTOS19, then edit mounted paths.\nPreparation reads individual archive entries with a bounded CPU process pool;\nit never extracts the original image tree or creates a combined archive.\n\nFast mode uses reduced JPEG decoding (minimum side 768 when supported), finds\ncrop bounds on a 512-pixel preview, and saves unenhanced 448x448 JPEGs at quality\n95. Both model resolutions use the same cached crop. Ben Graham/CLAHE are not\nenabled in this experiment. Original dimensions remain in the audit manifest.\n\nprepare_workers controls preparation independently of training workers. It is\ncapped by CPU affinity and available RAM. Spawned workers import only CPU code.\nEvery 50 completed images, SQLite checkpoints commit; interruptions lose at\nmost that batch of scan records, which is safe to repeat. Input/config changes\ninvalidate individual records. Unexpected missing/corrupt images still fail.\n\nFast exact hashes describe reduced RGB pixels, not native full-resolution RGB.\nPatient grouping, perceptual duplicate search, original train/test membership,\nand APTOS quarantine remain enabled. Set prepare_hash_mode='original_rgb' to\nretain strict original RGB hash semantics at the cost of full decoding/hashing.\nPreview cropping/reduced decoding/JPEG caching change input pixels, so use this\nnew experiment directory and retrain; old manifests/checkpoints are incompatible.\n\nRun RUN_STAGE='prepare' first, save its completed notebook output, then attach\nthat output in a new GPU session. Set restore_dir to the attached full folder\ncontaining manifest.csv, run_manifest.json, image_cache/, and scan SQLite files.\nRun train224, train448, predict_dev, router, benchmark, evaluate in separate\nsessions as needed. Restoring the complete folder reuses the scan and images;\ncache paths are relocated automatically. Save partial outputs too when pausing.\nThe disk reserve defaults to 1 GiB; disk exhaustion checkpoints completed work.\n\nTraining/prediction use cached images. End-to-end benchmarks still start from\noriginal inputs and include the same decode/crop/resize/JPEG preprocessing.\nMeasure timing and storage on Kaggle; no fixed speedup or accuracy is promised.\nAPTOS labels are reserved for final evaluation; smoke outputs are debug only.\n","metadata":{}},{"cell_type":"code","source":"from pathlib import Path\nimport os\nCFG = dict(\n    # Replace these paths with the paths shown in Kaggle's Input panel.\n    eye_images='/kaggle/input/competitions/diabetic-retinopathy-detection',\n    eye_train_csv=None,  # auto-find official trainLabels.csv or trainLabels.csv.zip\n    eye_test_csv=None,  # attach reference CSV path, or download this small file\n    confirmed_invalid_eye_images={},  # validate the originals before adding any exclusions\n    aptos_csv='/kaggle/input/competitions/aptos2019-blindness-detection/train.csv',\n    aptos_images='/kaggle/input/competitions/aptos2019-blindness-detection/train_images',\n    output='/kaggle/working/selective_dr_original_fast', restore_dir='/kaggle/input/datasets/khangcancode/processed-weight-for-selective-dr',\n    pretrained_weights=None, allow_download=True,\n    seed=42, epochs=15, lr=3e-4, weight_decay=1e-4,\n    effective_batch=32, max_microbatch=4, workers=0, smoke=False, smoke_groups=150,\n    # Separate CPU pool for preparation; bounded by CPU affinity and available RAM.\n    prepare_workers=min(8, os.cpu_count() or 1), prepare_resume=True,\n    prepare_checkpoint_every=50, prepare_decode_min_side=768, prepare_crop_preview=512,\n    # Fast hashes describe reduced RGB pixels; original_rgb retains strict native hashes.\n    prepare_hash_mode='decoded_rgb', prepare_cache=True,\n    prepare_cache_size=448, prepare_jpeg_quality=95, prepare_disk_reserve_gb=1.,\n    bootstrap=1000, benchmark_images=100, warmup=10,\n    near_distance=4, budgets=[.10, .25, .50, .75],\n)\nRUN_STAGE = 'all'  # prepare, train224, train448, predict_dev, router, benchmark, evaluate, all\n","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Imports, reproducibility, and artifact identity\nKaggle supplies these packages. No automatic package upgrades are performed.\n","metadata":{}},{"cell_type":"code","source":"import os, json, random, time, hashlib, shutil, platform, gc, tempfile\nimport numpy as np\nimport pandas as pd\nimport cv2\nfrom PIL import Image\nimport matplotlib.pyplot as plt\nimport torch\nfrom torch import nn\nfrom torch.utils.data import Dataset, DataLoader\nfrom torchvision import models, transforms\nimport torchvision\nimport sklearn\nfrom sklearn.model_selection import StratifiedGroupKFold\nfrom sklearn.pipeline import make_pipeline\nfrom sklearn.preprocessing import StandardScaler\nfrom sklearn.linear_model import Ridge\nfrom sklearn.metrics import cohen_kappa_score, confusion_matrix\nimport joblib\n\n\"\"\"Original EyePACS archive access for the self-contained Kaggle notebook.\n\nMounted multipart ZIP files are exposed as one seekable stream. Only requested\nentries are decompressed; no concatenated archive or extracted image tree is made.\n\"\"\"\nimport bisect\nimport contextlib\nimport hashlib\nimport io\nimport json\nimport os\nfrom functools import lru_cache\nfrom pathlib import Path\nimport urllib.request\nimport zipfile\n\nimport numpy as np\nimport pandas as pd\nfrom PIL import Image\n\n\nclass MultipartFile(io.RawIOBase):\n    def __init__(self, paths):\n        super().__init__()\n        self.paths = [Path(p) for p in paths]\n        self.offsets = [0]\n        for path in self.paths:\n            self.offsets.append(self.offsets[-1] + path.stat().st_size)\n        self.position = 0\n        self.handle = None\n        self.part = None\n\n    def readable(self):\n        return True\n\n    def seekable(self):\n        return True\n\n    def tell(self):\n        return self.position\n\n    def seek(self, offset, whence=io.SEEK_SET):\n        target = (offset if whence == io.SEEK_SET else\n                  self.position + offset if whence == io.SEEK_CUR else\n                  self.offsets[-1] + offset if whence == io.SEEK_END else None)\n        if target is None or target < 0:\n            raise ValueError('Invalid archive seek')\n        self.position = target\n        return target\n\n    def read(self, size=-1):\n        if self.closed:\n            raise ValueError('Archive is closed')\n        remaining = max(0, self.offsets[-1] - self.position)\n        size = remaining if size is None or size < 0 else min(size, remaining)\n        pieces = []\n        while size:\n            index = bisect.bisect_right(self.offsets, self.position) - 1\n            if index != self.part:\n                if self.handle:\n                    self.handle.close()\n                self.handle = self.paths[index].open('rb')\n                self.part = index\n            self.handle.seek(self.position - self.offsets[index])\n            count = min(size, self.offsets[index + 1] - self.position)\n            data = self.handle.read(count)\n            if len(data) != count:\n                raise OSError(f'Short archive read: {self.paths[index]}')\n            pieces.append(data)\n            self.position += count\n            size -= count\n        return b''.join(pieces)\n\n    def close(self):\n        if self.handle:\n            self.handle.close()\n            self.handle = None\n        super().close()\n\n\ndef archive_parts(first):\n    first = Path(first)\n    if first.name.endswith('.zip.001'):\n        base = first.name[:-4]\n        parts = sorted((p for p in first.parent.glob(base + '.*')\n                        if p.name.rsplit('.', 1)[-1].isdigit()),\n                       key=lambda p: int(p.name.rsplit('.', 1)[-1]))\n        numbers = [int(p.name.rsplit('.', 1)[-1]) for p in parts]\n        if not numbers or numbers != list(range(1, len(parts) + 1)):\n            raise ValueError(f'Missing/nonconsecutive multipart archive files: {first}')\n        expected = {'train.zip': 5, 'test.zip': 7}.get(base)\n        if expected and len(parts) != expected:\n            raise ValueError(f'{base}: expected {expected} original competition parts; found {len(parts)}')\n        return parts\n    return [first]\n\n\nclass Archive:\n    def __init__(self, first):\n        self.stream = MultipartFile(archive_parts(first))\n        try:\n            self.zip = zipfile.ZipFile(self.stream)\n        except Exception as error:\n            self.stream.close()\n            raise ValueError(f'Cannot open original archive {first}: {error}. '\n                             'Check all mounted parts; this loader supports concatenated ZIP byte streams.') from error\n\n    def __del__(self):\n        if hasattr(self, 'zip'):\n            self.zip.close()\n        if hasattr(self, 'stream'):\n            self.stream.close()\n\n\n@lru_cache(maxsize=8)\ndef _cached_archive_for_process(first, pid):\n    return Archive(first)\n\n\ndef cached_archive(first):\n    # DataLoader workers need independent file descriptions after a fork.\n    return _cached_archive_for_process(str(first), os.getpid())\n\n\ndef exact_rgb_hash(rgb):\n    digest = hashlib.sha256()\n    digest.update(memoryview(rgb))\n    digest.update(str(rgb.shape).encode())\n    return digest.hexdigest()\n\n\n@contextlib.contextmanager\ndef open_image(location):\n    \"\"\"One PIL image at a time, from a plain file or one ZIP entry.\"\"\"\n    location = str(location)\n    if location.startswith('zip:'):\n        archive, member = json.loads(location[4:])\n        with cached_archive(archive).zip.open(member) as stream:\n            with Image.open(stream) as image:\n                yield image\n    else:\n        with Image.open(location) as image:\n            yield image\n\n\ndef index_images(folder, dataset):\n    folder = Path(folder)\n    if not folder.is_dir():\n        raise FileNotFoundError(f'Attach the input dataset and check this path: {folder}')\n    files = {}\n    def add(image_id, location):\n        if image_id in files:\n            raise ValueError(f'Ambiguous image stem {image_id}')\n        files[image_id] = location\n    extracted = [folder / split for split in ('train', 'test') if (folder / split).is_dir()]\n    roots = extracted if dataset == 'eyepacs' and extracted else [folder]\n    for root in roots:\n        for path in root.rglob('*'):\n            if path.is_file() and path.suffix.lower() in ('.jpg', '.jpeg', '.png'):\n                add(path.stem, str(path))\n    if dataset == 'eyepacs':\n        for split in ('train', 'test'):\n            if (folder / split).is_dir():\n                continue\n            archives = list(folder.rglob(split + '.zip')) + list(folder.rglob(split + '.zip.001'))\n            if len(archives) > 1:\n                raise ValueError(f'Ambiguous original {split} archive: {archives}')\n            for archive in archives:\n                handle = cached_archive(str(archive))\n                count = 0\n                for member in handle.zip.infolist():\n                    path = Path(member.filename)\n                    if not member.is_dir() and path.suffix.lower() in ('.jpg', '.jpeg', '.png'):\n                        add(path.stem, 'zip:' + json.dumps([str(archive), member.filename], separators=(',', ':')))\n                        count += 1\n                print(f'Indexed {split}: {count} images directly from mounted archive.')\n    if not files:\n        raise ValueError(f'No image files or original train/test archives found in {folder}')\n    return files\n\n\ndef original_train_csv():\n    if CFG.get('eye_train_csv'):\n        return Path(CFG['eye_train_csv'])\n    root = Path(CFG['eye_images'])\n    choices = list(root.rglob('trainLabels.csv')) + list(root.rglob('trainLabels.csv.zip'))\n    if len(choices) != 1:\n        raise ValueError(f'Expected one official trainLabels.csv or trainLabels.csv.zip in {root}; found {choices}')\n    return choices[0]\n\n\ndef original_test_csv():\n    if CFG.get('eye_test_csv'):\n        return Path(CFG['eye_test_csv'])\n    path = OUT / 'reference_retinopathy_solution.csv'\n    if not path.exists():\n        if not CFG['allow_download']:\n            raise RuntimeError('Attach retinopathy_solution.csv and set eye_test_csv, or enable '\n                               'Kaggle Internet and allow_download for the small test-label file.')\n        url = ('https://storage.googleapis.com/kaggle-forum-message-attachments/'\n               '90528/2877/retinopathy_solution.csv')\n        with urllib.request.urlopen(url, timeout=60) as response:\n            content = response.read(5_000_001)\n        if len(content) > 5_000_000:\n            raise ValueError('Unexpected size for competition test-label CSV')\n        expected = 'a6af3d66804e81887da89b9261bcfb7da4f5bacedb070f1b5db83fa879ab60c9'\n        if hashlib.sha256(content).hexdigest() != expected:\n            raise ValueError('Competition test-label checksum differs from verified reference; inspect the source')\n        path.write_bytes(content)\n    return path\n\n\ndef original_eye_labels():\n    train = read_eye_labels(original_train_csv()).assign(source_split='original_train')\n    test = read_eye_labels(original_test_csv()).assign(source_split='original_test')\n    overlap = set(train.image) & set(test.image)\n    if overlap:\n        raise ValueError(f'Original train/test CSVs overlap: {sorted(overlap)[:10]}')\n    print(f'Official EyePACS labels: {len(train)} original train; {len(test)} original test.')\n    pd.DataFrame(columns=['image_id', 'original_grade', 'source_split', 'reason', 'error']).to_csv(\n        OUT / 'eyepacs_excluded_samples.csv', index=False)\n    return pd.concat([train, test], ignore_index=True)\n\nPREPROCESSING_SOURCES = {'original_eyepacs_io.py': '\"\"\"Original EyePACS archive access for the self-contained Kaggle notebook.\\n\\nMounted multipart ZIP files are exposed as one seekable stream. Only requested\\nentries are decompressed; no concatenated archive or extracted image tree is made.\\n\"\"\"\\nimport bisect\\nimport contextlib\\nimport hashlib\\nimport io\\nimport json\\nimport os\\nfrom functools import lru_cache\\nfrom pathlib import Path\\nimport urllib.request\\nimport zipfile\\n\\nimport numpy as np\\nimport pandas as pd\\nfrom PIL import Image\\n\\n\\nclass MultipartFile(io.RawIOBase):\\n    def __init__(self, paths):\\n        super().__init__()\\n        self.paths = [Path(p) for p in paths]\\n        self.offsets = [0]\\n        for path in self.paths:\\n            self.offsets.append(self.offsets[-1] + path.stat().st_size)\\n        self.position = 0\\n        self.handle = None\\n        self.part = None\\n\\n    def readable(self):\\n        return True\\n\\n    def seekable(self):\\n        return True\\n\\n    def tell(self):\\n        return self.position\\n\\n    def seek(self, offset, whence=io.SEEK_SET):\\n        target = (offset if whence == io.SEEK_SET else\\n                  self.position + offset if whence == io.SEEK_CUR else\\n                  self.offsets[-1] + offset if whence == io.SEEK_END else None)\\n        if target is None or target < 0:\\n            raise ValueError(\\'Invalid archive seek\\')\\n        self.position = target\\n        return target\\n\\n    def read(self, size=-1):\\n        if self.closed:\\n            raise ValueError(\\'Archive is closed\\')\\n        remaining = max(0, self.offsets[-1] - self.position)\\n        size = remaining if size is None or size < 0 else min(size, remaining)\\n        pieces = []\\n        while size:\\n            index = bisect.bisect_right(self.offsets, self.position) - 1\\n            if index != self.part:\\n                if self.handle:\\n                    self.handle.close()\\n                self.handle = self.paths[index].open(\\'rb\\')\\n                self.part = index\\n            self.handle.seek(self.position - self.offsets[index])\\n            count = min(size, self.offsets[index + 1] - self.position)\\n            data = self.handle.read(count)\\n            if len(data) != count:\\n                raise OSError(f\\'Short archive read: {self.paths[index]}\\')\\n            pieces.append(data)\\n            self.position += count\\n            size -= count\\n        return b\\'\\'.join(pieces)\\n\\n    def close(self):\\n        if self.handle:\\n            self.handle.close()\\n            self.handle = None\\n        super().close()\\n\\n\\ndef archive_parts(first):\\n    first = Path(first)\\n    if first.name.endswith(\\'.zip.001\\'):\\n        base = first.name[:-4]\\n        parts = sorted((p for p in first.parent.glob(base + \\'.*\\')\\n                        if p.name.rsplit(\\'.\\', 1)[-1].isdigit()),\\n                       key=lambda p: int(p.name.rsplit(\\'.\\', 1)[-1]))\\n        numbers = [int(p.name.rsplit(\\'.\\', 1)[-1]) for p in parts]\\n        if not numbers or numbers != list(range(1, len(parts) + 1)):\\n            raise ValueError(f\\'Missing/nonconsecutive multipart archive files: {first}\\')\\n        expected = {\\'train.zip\\': 5, \\'test.zip\\': 7}.get(base)\\n        if expected and len(parts) != expected:\\n            raise ValueError(f\\'{base}: expected {expected} original competition parts; found {len(parts)}\\')\\n        return parts\\n    return [first]\\n\\n\\nclass Archive:\\n    def __init__(self, first):\\n        self.stream = MultipartFile(archive_parts(first))\\n        try:\\n            self.zip = zipfile.ZipFile(self.stream)\\n        except Exception as error:\\n            self.stream.close()\\n            raise ValueError(f\\'Cannot open original archive {first}: {error}. \\'\\n                             \\'Check all mounted parts; this loader supports concatenated ZIP byte streams.\\') from error\\n\\n    def __del__(self):\\n        if hasattr(self, \\'zip\\'):\\n            self.zip.close()\\n        if hasattr(self, \\'stream\\'):\\n            self.stream.close()\\n\\n\\n@lru_cache(maxsize=8)\\ndef _cached_archive_for_process(first, pid):\\n    return Archive(first)\\n\\n\\ndef cached_archive(first):\\n    # DataLoader workers need independent file descriptions after a fork.\\n    return _cached_archive_for_process(str(first), os.getpid())\\n\\n\\ndef exact_rgb_hash(rgb):\\n    digest = hashlib.sha256()\\n    digest.update(memoryview(rgb))\\n    digest.update(str(rgb.shape).encode())\\n    return digest.hexdigest()\\n\\n\\n@contextlib.contextmanager\\ndef open_image(location):\\n    \"\"\"One PIL image at a time, from a plain file or one ZIP entry.\"\"\"\\n    location = str(location)\\n    if location.startswith(\\'zip:\\'):\\n        archive, member = json.loads(location[4:])\\n        with cached_archive(archive).zip.open(member) as stream:\\n            with Image.open(stream) as image:\\n                yield image\\n    else:\\n        with Image.open(location) as image:\\n            yield image\\n\\n\\ndef index_images(folder, dataset):\\n    folder = Path(folder)\\n    if not folder.is_dir():\\n        raise FileNotFoundError(f\\'Attach the input dataset and check this path: {folder}\\')\\n    files = {}\\n    def add(image_id, location):\\n        if image_id in files:\\n            raise ValueError(f\\'Ambiguous image stem {image_id}\\')\\n        files[image_id] = location\\n    extracted = [folder / split for split in (\\'train\\', \\'test\\') if (folder / split).is_dir()]\\n    roots = extracted if dataset == \\'eyepacs\\' and extracted else [folder]\\n    for root in roots:\\n        for path in root.rglob(\\'*\\'):\\n            if path.is_file() and path.suffix.lower() in (\\'.jpg\\', \\'.jpeg\\', \\'.png\\'):\\n                add(path.stem, str(path))\\n    if dataset == \\'eyepacs\\':\\n        for split in (\\'train\\', \\'test\\'):\\n            if (folder / split).is_dir():\\n                continue\\n            archives = list(folder.rglob(split + \\'.zip\\')) + list(folder.rglob(split + \\'.zip.001\\'))\\n            if len(archives) > 1:\\n                raise ValueError(f\\'Ambiguous original {split} archive: {archives}\\')\\n            for archive in archives:\\n                handle = cached_archive(str(archive))\\n                count = 0\\n                for member in handle.zip.infolist():\\n                    path = Path(member.filename)\\n                    if not member.is_dir() and path.suffix.lower() in (\\'.jpg\\', \\'.jpeg\\', \\'.png\\'):\\n                        add(path.stem, \\'zip:\\' + json.dumps([str(archive), member.filename], separators=(\\',\\', \\':\\')))\\n                        count += 1\\n                print(f\\'Indexed {split}: {count} images directly from mounted archive.\\')\\n    if not files:\\n        raise ValueError(f\\'No image files or original train/test archives found in {folder}\\')\\n    return files\\n\\n\\ndef original_train_csv():\\n    if CFG.get(\\'eye_train_csv\\'):\\n        return Path(CFG[\\'eye_train_csv\\'])\\n    root = Path(CFG[\\'eye_images\\'])\\n    choices = list(root.rglob(\\'trainLabels.csv\\')) + list(root.rglob(\\'trainLabels.csv.zip\\'))\\n    if len(choices) != 1:\\n        raise ValueError(f\\'Expected one official trainLabels.csv or trainLabels.csv.zip in {root}; found {choices}\\')\\n    return choices[0]\\n\\n\\ndef original_test_csv():\\n    if CFG.get(\\'eye_test_csv\\'):\\n        return Path(CFG[\\'eye_test_csv\\'])\\n    path = OUT / \\'reference_retinopathy_solution.csv\\'\\n    if not path.exists():\\n        if not CFG[\\'allow_download\\']:\\n            raise RuntimeError(\\'Attach retinopathy_solution.csv and set eye_test_csv, or enable \\'\\n                               \\'Kaggle Internet and allow_download for the small test-label file.\\')\\n        url = (\\'https://storage.googleapis.com/kaggle-forum-message-attachments/\\'\\n               \\'90528/2877/retinopathy_solution.csv\\')\\n        with urllib.request.urlopen(url, timeout=60) as response:\\n            content = response.read(5_000_001)\\n        if len(content) > 5_000_000:\\n            raise ValueError(\\'Unexpected size for competition test-label CSV\\')\\n        expected = \\'a6af3d66804e81887da89b9261bcfb7da4f5bacedb070f1b5db83fa879ab60c9\\'\\n        if hashlib.sha256(content).hexdigest() != expected:\\n            raise ValueError(\\'Competition test-label checksum differs from verified reference; inspect the source\\')\\n        path.write_bytes(content)\\n    return path\\n\\n\\ndef original_eye_labels():\\n    train = read_eye_labels(original_train_csv()).assign(source_split=\\'original_train\\')\\n    test = read_eye_labels(original_test_csv()).assign(source_split=\\'original_test\\')\\n    overlap = set(train.image) & set(test.image)\\n    if overlap:\\n        raise ValueError(f\\'Original train/test CSVs overlap: {sorted(overlap)[:10]}\\')\\n    print(f\\'Official EyePACS labels: {len(train)} original train; {len(test)} original test.\\')\\n    pd.DataFrame(columns=[\\'image_id\\', \\'original_grade\\', \\'source_split\\', \\'reason\\', \\'error\\']).to_csv(\\n        OUT / \\'eyepacs_excluded_samples.csv\\', index=False)\\n    return pd.concat([train, test], ignore_index=True)\\n', 'original_preprocessing.py': '\"\"\"CPU-only, importable workers for original EyePACS preparation.\\n\\nThe notebook embeds this module and its archive reader, so spawned processes do\\nnot import the notebook, torch, or initialize CUDA. No image arrays cross IPC.\\n\"\"\"\\nimport concurrent.futures as futures\\nimport hashlib\\nimport io\\nimport json\\nimport math\\nimport multiprocessing\\nimport os\\nfrom pathlib import Path\\nimport shutil\\nimport sqlite3\\nimport time\\n\\nimport cv2\\nimport numpy as np\\nimport pandas as pd\\nfrom PIL import Image, ImageFile\\n\\nfrom original_eyepacs_io import (_cached_archive_for_process, archive_parts, exact_rgb_hash,\\n                                index_images, open_image)\\n\\nCODE_VERSION = \\'parallel-original-v1\\'\\n\\n\\ndef worker_init():\\n    cv2.setNumThreads(1)\\n    ImageFile.LOAD_TRUNCATED_IMAGES = False\\n\\n\\ndef settings(cfg):\\n    options = dict(version=CODE_VERSION,\\n                   decode_min_side=int(cfg.get(\\'prepare_decode_min_side\\', 768)),\\n                   crop_preview=int(cfg.get(\\'prepare_crop_preview\\', 512)),\\n                   hash_mode=cfg.get(\\'prepare_hash_mode\\', \\'original_rgb\\'),\\n                   cache=bool(cfg.get(\\'prepare_cache\\', False)),\\n                   cache_size=int(cfg.get(\\'prepare_cache_size\\', 448)),\\n                   jpeg_quality=int(cfg.get(\\'prepare_jpeg_quality\\', 95)))\\n    if options[\\'hash_mode\\'] not in (\\'original_rgb\\', \\'decoded_rgb\\'):\\n        raise ValueError(\\'prepare_hash_mode must be original_rgb or decoded_rgb\\')\\n    if min(options[\\'crop_preview\\'], options[\\'cache_size\\']) < 1 or options[\\'decode_min_side\\'] < 0:\\n        raise ValueError(\\'Invalid preparation dimensions\\')\\n    if not 1 <= options[\\'jpeg_quality\\'] <= 100:\\n        raise ValueError(\\'prepare_jpeg_quality must be in 1..100\\')\\n    return options\\n\\n\\ndef read_rgb(location, min_side=768):\\n    with open_image(location) as image:\\n        width, height = image.size\\n        reduction = 1\\n        if min_side > 0 and image.format == \\'JPEG\\':\\n            while min(width, height) // (reduction * 2) >= min_side and reduction < 8:\\n                reduction *= 2\\n            if reduction > 1:\\n                image.draft(\\'RGB\\', (math.ceil(width / reduction), math.ceil(height / reduction)))\\n        rgb = np.asarray(image.convert(\\'RGB\\'))\\n    return rgb, width, height\\n\\n\\ndef crop_bounds(rgb, preview):\\n    height, width = rgb.shape[:2]\\n    scale = min(1., preview / max(height, width))\\n    small = cv2.resize(rgb, (max(1, round(width * scale)), max(1, round(height * scale))),\\n                       interpolation=cv2.INTER_AREA) if scale < 1 else rgb\\n    mask = small.max(axis=2) > 10\\n    yy = np.flatnonzero(mask.any(axis=1)); xx = np.flatnonzero(mask.any(axis=0))\\n    if not len(xx):\\n        return 0, height, 0, width\\n    sy, sx = height / small.shape[0], width / small.shape[1]\\n    return (max(0, math.floor(yy[0] * sy)), min(height, math.ceil((yy[-1] + 1) * sy)),\\n            max(0, math.floor(xx[0] * sx)), min(width, math.ceil((xx[-1] + 1) * sx)))\\n\\n\\ndef image_pixels(location, options, audit=True):\\n    # Strict mode intentionally retains the cost of full RGB decoding/hashing.\\n    full_decode = options[\\'hash_mode\\'] == \\'original_rgb\\'\\n    rgb, width, height = read_rgb(location, 0 if full_decode else options[\\'decode_min_side\\'])\\n    exact = exact_rgb_hash(rgb) if audit else None\\n    y0, y1, x0, x1 = crop_bounds(rgb, options[\\'crop_preview\\'])\\n    crop = rgb[y0:y1, x0:x1]\\n    geometry = dict(width=width, height=height,\\n                    crop_width=round((x1-x0) * width / rgb.shape[1]),\\n                    crop_height=round((y1-y0) * height / rgb.shape[0]),\\n                    decoded_width=rgb.shape[1], decoded_height=rgb.shape[0])\\n    return crop, exact, geometry\\n\\n\\ndef jpeg_bytes(crop, options):\\n    resized = cv2.resize(crop, (options[\\'cache_size\\'], options[\\'cache_size\\']), interpolation=cv2.INTER_AREA)\\n    ok, encoded = cv2.imencode(\\'.jpg\\', cv2.cvtColor(resized, cv2.COLOR_RGB2BGR),\\n                              [cv2.IMWRITE_JPEG_QUALITY, options[\\'jpeg_quality\\']])\\n    if not ok:\\n        raise OSError(\\'JPEG encoding failed\\')\\n    return encoded.tobytes()\\n\\n\\ndef model_rgb(location, cfg):\\n    \"\"\"Raw-image inference path matching cached training pixels, including JPEG.\"\"\"\\n    options = settings(cfg)\\n    crop, _, _ = image_pixels(location, options, audit=False)\\n    if not options[\\'cache\\']:\\n        return crop\\n    with Image.open(io.BytesIO(jpeg_bytes(crop, options))) as image:\\n        return np.asarray(image.convert(\\'RGB\\'))\\n\\n\\ndef process_image(task):\\n    image_id, grade, location, dataset, options, destination = task\\n    try:\\n        crop, exact, geometry = image_pixels(location, options)\\n        gray = cv2.cvtColor(crop, cv2.COLOR_RGB2GRAY)\\n        d = cv2.dct(cv2.resize(gray, (32, 32)).astype(np.float32))[:8, :8].ravel()\\n        bits = d > np.median(d[1:]); bits[0] = False\\n        phash = sum(int(b) << i for i, b in enumerate(bits))\\n        if dataset == \\'eyepacs\\':\\n            patient, sep, eye = image_id.rpartition(\\'_\\')\\n            if not sep or eye not in (\\'left\\', \\'right\\'):\\n                raise ValueError(\\'Expected original patient_left/right filename\\')\\n        else:\\n            patient = image_id\\n        row = dict(image_id=image_id, grade=int(grade), dataset=dataset, source_path=location,\\n                   patient_id=dataset+\\':\\'+patient, **geometry,\\n                   insufficient_detail=min(geometry[\\'crop_width\\'], geometry[\\'crop_height\\']) < 448,\\n                   exact=exact, exact_hash_scope=options[\\'hash_mode\\'], phash_hex=hex(phash),\\n                   cache_file=\\'\\', cache_sha256=\\'\\', cache_bytes=0)\\n        if options[\\'cache\\']:\\n            content = jpeg_bytes(crop, options)\\n            target = Path(destination); target.parent.mkdir(parents=True, exist_ok=True)\\n            temporary = target.with_suffix(\\'.jpg.tmp\\')\\n            temporary.write_bytes(content); os.replace(temporary, target)\\n            row.update(cache_sha256=hashlib.sha256(content).hexdigest(), cache_bytes=len(content))\\n        return row, None\\n    except Exception as error:\\n        return None, str(error)\\n\\n\\ndef worker_count(cfg):\\n    try:\\n        cpus = len(os.sched_getaffinity(0))\\n    except AttributeError:\\n        cpus = os.cpu_count() or 1\\n    count = max(1, min(int(cfg.get(\\'prepare_workers\\', min(8, cpus))), cpus))\\n    try:\\n        import psutil\\n        per_worker = (768 if cfg.get(\\'prepare_hash_mode\\', \\'original_rgb\\') == \\'original_rgb\\' else 384) * 2**20\\n        count = min(count, max(1, int(psutil.virtual_memory().available * .4 // per_worker)))\\n    except ImportError:\\n        count = min(count, 4)\\n    return count\\n\\n\\ndef source_stamp(location):\\n    if location.startswith(\\'zip:\\'):\\n        archive, _ = json.loads(location[4:])\\n        paths = archive_parts(archive)\\n    else:\\n        paths = [Path(location)]\\n    return [(str(p), p.stat().st_size, p.stat().st_mtime_ns) for p in paths]\\n\\n\\ndef scan(csv, folder, dataset, cfg, out):\\n    \"\"\"Parallel, bounded, deterministic scan with durable per-image checkpoints.\"\"\"\\n    out = Path(out); out.mkdir(parents=True, exist_ok=True)\\n    labels = csv.copy() if isinstance(csv, pd.DataFrame) else pd.read_csv(csv)\\n    idcol, gradecol = (\\'image\\', \\'level\\') if dataset == \\'eyepacs\\' else (\\'id_code\\', \\'diagnosis\\')\\n    assert {idcol, gradecol}.issubset(labels.columns)\\n    assert labels[idcol].is_unique, \\'Duplicate image IDs in labels\\'\\n    assert labels[gradecol].notna().all() and labels[gradecol].isin(range(5)).all()\\n    options = settings(cfg); files = index_images(folder, dataset)\\n    resume = bool(cfg.get(\\'prepare_resume\\', True))\\n    db = sqlite3.connect(out / f\\'{dataset}_scan.sqlite3\\') if resume else None\\n    if db:\\n        db.execute(\\'CREATE TABLE IF NOT EXISTS records (key TEXT PRIMARY KEY, result TEXT NOT NULL)\\')\\n    rows = {}; errors = []; exclusions = []; tasks = []; stamps = {}\\n    invalid = cfg.get(\\'confirmed_invalid_eye_images\\', {}) if dataset == \\'eyepacs\\' else {}\\n    checkpoint_every = max(1, int(cfg.get(\\'prepare_checkpoint_every\\', 50)))\\n    stats = dict(dataset=dataset, images=len(labels), reused=0, processed=0,\\n                 workers=worker_count(cfg), hash_scope=options[\\'hash_mode\\'])\\n    started = time.perf_counter(); last_print = started\\n    try:\\n        for index, (image_id, grade) in enumerate(labels[[idcol, gradecol]].itertuples(index=False, name=None)):\\n            image_id = str(image_id); location = files.get(image_id)\\n            if location is None:\\n                errors.append(dict(image_id=image_id, error=\\'missing\\')); continue\\n            stamp_key = location\\n            # All members of one archive share the same physical input signature.\\n            if location.startswith(\\'zip:\\'):\\n                stamp_key = json.loads(location[4:])[0]\\n            if stamp_key not in stamps:\\n                stamps[stamp_key] = source_stamp(location)\\n            payload = [image_id, int(grade), location, stamps[stamp_key], options]\\n            key = hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()\\n            relative = f\\'image_cache/{dataset}/{key[:32]}.jpg\\' if options[\\'cache\\'] else \\'\\'\\n            previous = db.execute(\\'SELECT result FROM records WHERE key=?\\', (key,)).fetchone() if db else None\\n            if previous:\\n                row = json.loads(previous[0])\\n                target = out / relative\\n                if not options[\\'cache\\'] or (target.is_file() and target.stat().st_size == row[\\'cache_bytes\\']):\\n                    rows[index] = row; stats[\\'reused\\'] += 1; continue\\n            tasks.append((index, key, relative, (image_id, int(grade), location, dataset, options, str(out / relative))))\\n        print(f\"{dataset}: {stats[\\'reused\\']} reused; {len(tasks)} pending; {stats[\\'workers\\']} CPU processes; \"\\n              f\"hash={options[\\'hash_mode\\']}; cache={options[\\'cache\\']}\", flush=True)\\n\\n        def accept(item, result):\\n            nonlocal last_print\\n            index, key, relative, task = item\\n            row, error = result\\n            if error is not None:\\n                image_id, grade = task[:2]\\n                if invalid.get(image_id) and invalid[image_id] in error:\\n                    source = labels.iloc[index]\\n                    exclusions.append(dict(image_id=image_id, original_grade=grade,\\n                        source_split=source.get(\\'source_split\\'), reason=\\'confirmed_truncated_image\\', error=error))\\n                else:\\n                    errors.append(dict(image_id=image_id, error=error))\\n            else:\\n                row[\\'cache_file\\'] = relative; rows[index] = row\\n                if db:\\n                    db.execute(\\'INSERT OR REPLACE INTO records VALUES (?, ?)\\', (key, json.dumps(row)))\\n            stats[\\'processed\\'] += 1\\n            if db and stats[\\'processed\\'] % checkpoint_every == 0:\\n                db.commit()\\n            now = time.perf_counter()\\n            if now-last_print >= 30 or stats[\\'processed\\'] == len(tasks):\\n                print(f\"{dataset}: {stats[\\'reused\\']+stats[\\'processed\\']}/{len(labels)} scanned; \"\\n                      f\"{stats[\\'processed\\']/max(.001,now-started):.2f} new images/s\", flush=True)\\n                last_print = now\\n\\n        def check_disk():\\n            if options[\\'cache\\'] and shutil.disk_usage(out).free < float(cfg.get(\\'prepare_disk_reserve_gb\\', 1.)) * 2**30:\\n                raise RuntimeError(\\'Image cache reached the disk reserve. Save this output and resume on storage with more space.\\')\\n\\n        if stats[\\'workers\\'] == 1:\\n            worker_init()\\n            for item in tasks:\\n                check_disk(); accept(item, process_image(item[3]))\\n        elif tasks:\\n            # Spawn avoids inheriting initialized torch/CUDA state from the notebook.\\n            context = multiprocessing.get_context(\\'spawn\\')\\n            with futures.ProcessPoolExecutor(max_workers=stats[\\'workers\\'], mp_context=context,\\n                                            initializer=worker_init) as executor:\\n                pending = {}; iterator = iter(tasks)\\n                def fill():\\n                    check_disk()\\n                    while len(pending) < stats[\\'workers\\'] * 2:\\n                        item = next(iterator, None)\\n                        if item is None:\\n                            break\\n                        pending[executor.submit(process_image, item[3])] = item\\n                fill()\\n                try:\\n                    while pending:\\n                        done, _ = futures.wait(pending, return_when=futures.FIRST_COMPLETED)\\n                        for future in done:\\n                            accept(pending.pop(future), future.result())\\n                        fill()\\n                finally:\\n                    for future in pending:\\n                        future.cancel()\\n    finally:\\n        if db:\\n            db.commit(); db.close()\\n        _cached_archive_for_process.cache_clear()\\n        stats[\\'elapsed_seconds\\'] = time.perf_counter()-started\\n        (out / f\\'{dataset}_scan_timing.json\\').write_text(json.dumps(stats, indent=2))\\n    pd.DataFrame(errors, columns=[\\'image_id\\', \\'error\\']).to_csv(out / f\\'{dataset}_invalid.csv\\', index=False)\\n    if dataset == \\'eyepacs\\':\\n        audit_path = out / \\'eyepacs_excluded_samples.csv\\'\\n        audit = pd.read_csv(audit_path) if audit_path.exists() else pd.DataFrame(columns=[\\n            \\'image_id\\', \\'original_grade\\', \\'source_split\\', \\'reason\\', \\'error\\'])\\n        audit = audit[audit.reason != \\'confirmed_truncated_image\\']\\n        if exclusions:\\n            audit = pd.concat([audit, pd.DataFrame(exclusions)], ignore_index=True)\\n        audit.to_csv(audit_path, index=False)\\n    if errors:\\n        raise RuntimeError(f\\'{dataset}: {len(errors)} invalid images; first errors: {errors[:10]}; inspect {dataset}_invalid.csv.\\')\\n    if not rows:\\n        raise ValueError(f\\'No labeled images found for {dataset}\\')\\n    frame = pd.DataFrame([rows[i] for i in sorted(rows)])\\n    frame[\\'path\\'] = [str(out / r.cache_file) if r.cache_file else r.source_path for r in frame.itertuples()]\\n    if \\'source_split\\' in labels:\\n        frame[\\'source_split\\'] = frame.image_id.map(labels.assign(**{idcol: labels[idcol].astype(str)}).set_index(idcol).source_split)\\n    return frame\\n'}\n\ndef preprocessing_runtime():\n    import importlib, sys\n    runtime = OUT / '_preprocess_runtime'\n    runtime.mkdir(parents=True, exist_ok=True)\n    changed = False\n    for name, source in PREPROCESSING_SOURCES.items():\n        path = runtime / name\n        if not path.exists() or path.read_text(encoding='utf-8') != source:\n            path.write_text(source, encoding='utf-8')\n            changed = True\n    if str(runtime) not in sys.path:\n        sys.path.insert(0, str(runtime))\n    if changed:\n        sys.modules.pop('original_preprocessing', None)\n        old_io = sys.modules.pop('original_eyepacs_io', None)\n        if old_io is not None:\n            old_io._cached_archive_for_process.cache_clear()\n        importlib.invalidate_caches()\n    return importlib.import_module('original_preprocessing')\n\ndef load_fundus(path):\n    # Paths in the manifest refer to already cropped and resized cache JPEGs.\n    if 'image_cache' in Path(str(path)).parts:\n        with open_image(path) as image:\n            return np.asarray(image.convert('RGB'))\n    return preprocessing_runtime().model_rgb(path, CFG)\n\ndef resolve_cache_paths(frame):\n    if 'cache_file' in frame.columns:\n        for index in frame.index[frame.cache_file.fillna('').ne('')]:\n            relative = Path(frame.at[index, 'cache_file'])\n            target = (OUT / relative).resolve()\n            if not target.is_relative_to(OUT.resolve()):\n                raise ValueError('Cache path escapes output directory')\n            if not target.is_file():\n                raise RuntimeError(f'Missing image cache {target}; restore the entire full folder, including image_cache.')\n            frame.at[index, 'path'] = str(target)\n    return frame\n\n\nOUT = Path(CFG['output']) / ('smoke' if CFG['smoke'] else 'full')\nOUT.mkdir(parents=True, exist_ok=True)\n\ntry:\n    import psutil\n    ram = psutil.virtual_memory()\n    print(f'RAM: {ram.total / 2**30:.1f} GiB total; {ram.available / 2**30:.1f} GiB available.')\nexcept ImportError:\n    print('RAM reporting unavailable; inspect Kaggle session resource meters.')\ndisk = shutil.disk_usage(OUT)\nprint(f'Working disk: {disk.free / 2**30:.1f} GiB free. Original archives are read from mounted inputs.')\nprint('Preparation uses a bounded CPU process pool and resumable image checkpoints. Cache JPEGs are saved in image_cache; original archives remain mounted.')\nif CFG['restore_dir']:\n    source = Path(CFG['restore_dir'])\n    assert source.is_dir(), f'Missing restore directory: {source}'\n    for p in source.iterdir():\n        dest = OUT / p.name\n        if not dest.exists():\n            shutil.copytree(p, dest) if p.is_dir() else shutil.copy2(p, dest)\n\ndef seed_all(seed):\n    random.seed(seed); np.random.seed(seed); torch.manual_seed(seed)\n    if torch.cuda.is_available(): torch.cuda.manual_seed_all(seed)\n    torch.backends.cudnn.benchmark = False\nseed_all(CFG['seed'])\nDEVICE = torch.device('cuda' if torch.cuda.is_available() else 'cpu')\nif CFG['smoke']: CFG['epochs'] = 1; CFG['bootstrap'] = 20\nidentity = {k:v for k,v in CFG.items() if k not in ('restore_dir', 'output', 'benchmark_images', 'warmup', 'workers', 'prepare_workers', 'prepare_checkpoint_every', 'prepare_disk_reserve_gb', 'prepare_resume')}\nidentity['implementation'] = 4\nidentity['preprocessing_source'] = hashlib.sha256(json.dumps(PREPROCESSING_SOURCES, sort_keys=True).encode()).hexdigest()\nsignature = hashlib.sha256(json.dumps(identity, sort_keys=True).encode()).hexdigest()\nrunfile = OUT / 'run_manifest.json'\nif runfile.exists():\n    assert json.loads(runfile.read_text())['signature'] == signature, 'Configuration changed: use a new output directory.'\nelse:\n    runfile.write_text(json.dumps(dict(signature=signature, config=CFG, python=platform.python_version(),\n        torch=torch.__version__, torchvision=torchvision.__version__, sklearn=sklearn.__version__,\n        gpu=torch.cuda.get_device_name(0) if DEVICE.type=='cuda' else 'CPU',\n        limitations=['Single training seed', 'APTOS patient IDs unavailable',\n                      'Bootstrap intervals exclude training-seed variability']), indent=2))\n\ndef atomic_torch(obj, path):\n    tmp = Path(str(path)+'.tmp'); torch.save(obj, tmp); os.replace(tmp, path)\n\ndef require(name):\n    p = OUT/name\n    if not p.exists(): raise RuntimeError(f'Missing {name}: run the prerequisite stage or restore prior outputs.')\n    return p\n\ndef crop_retina(rgb):\n    mask = rgb.max(axis=2) > 10\n    yy = np.flatnonzero(mask.any(axis=1)); xx = np.flatnonzero(mask.any(axis=0))\n    return rgb[yy.min():yy.max()+1, xx.min():xx.max()+1] if len(xx) else rgb\n\ndef quality(rgb):\n    gray = cv2.cvtColor(rgb, cv2.COLOR_RGB2GRAY)\n    return [cv2.Laplacian(gray, cv2.CV_32F).var(), gray.mean()/255,\n            gray.std()/255, ((gray<=5)|(gray>=250)).mean()]\n\ndef phash(rgb):\n    gray = cv2.cvtColor(rgb, cv2.COLOR_RGB2GRAY)\n    d = cv2.dct(cv2.resize(gray, (32,32)).astype(np.float32))[:8,:8].ravel()\n    bits = d > np.median(d[1:]); bits[0] = False\n    return sum(int(b)<<i for i,b in enumerate(bits))\n\nclass UnionFind:\n    def __init__(self,n): self.p=list(range(n))\n    def find(self,x):\n        while self.p[x]!=x: self.p[x]=self.p[self.p[x]]; x=self.p[x]\n        return x\n    def union(self,a,b): self.p[self.find(a)]=self.find(b)\n\nclass BKTree:\n    # Hamming-distance search avoids a quadratic all-pairs scan.\n    def __init__(self): self.root=None\n    def add(self,h,i):\n        if self.root is None: self.root=[h,[i],{}]; return\n        node=self.root\n        while True:\n            d=(h^node[0]).bit_count()\n            if d==0: node[1].append(i); return\n            if d not in node[2]: node[2][d]=[h,[i],{}]; return\n            node=node[2][d]\n    def query(self,h,r):\n        pending=[] if self.root is None else [self.root]; result=[]\n        while pending:\n            node=pending.pop(); d=(h^node[0]).bit_count()\n            if d<=r: result.extend(node[1])\n            pending.extend(v for k,v in node[2].items() if d-r<=k<=d+r)\n        return result\n\ndef contact_sheet(frame, pairs, filename):\n    pairs=pairs[:24]\n    if not pairs: return\n    fig,axes=plt.subplots(len(pairs),2,figsize=(8,3*len(pairs)),squeeze=False)\n    for row,(a,b) in enumerate(pairs):\n        for col,i in enumerate((a,b)):\n            with open_image(frame.iloc[i].path) as image: axes[row,col].imshow(image.convert('RGB'))\n            axes[row,col].axis('off')\n            axes[row,col].set_title(str(frame.iloc[i].image_id))\n    fig.tight_layout(); fig.savefig(OUT/filename); plt.close(fig)\n\n","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Preparation and immutable grouped splits\nThe original EyePACS test CSV defines `internal_test` and is reserved for final\nevaluation. The original training CSV is split by patient/duplicate group into\n`backbone_train`, `checkpoint_selection`, `router_train`, and `policy_validation`\nusing 13/2/2/1 of 18 stratified folds (approximately 72.2%/11.1%/11.1%/5.6%).\nThe proportions apply to the eligible training portion, not the whole dataset.\n\nIf a patient/duplicate group occurs in both original partitions, its training\nimages are quarantined; the original test images remain in `internal_test`.\n`eyepacs_split_membership.csv` records every image's original partition and\nassigned split, including quarantined images. No physical image copies are needed.\n\nPerceptual matches are conservative leakage groups, not proven duplicates.\nCross-dataset candidates within the configured Hamming distance are quarantined.\nConflicting duplicate labels are reported, retained, and grouped together.\n","metadata":{}},{"cell_type":"code","source":"def read_eye_labels(csv, require_grades=True):\n    \"\"\"Normalize CSV headers and filename extensions without changing patient IDs.\"\"\"\n    labels = pd.read_csv(csv, dtype=str)\n    labels.columns = labels.columns.str.strip().str.lower()\n    if not labels.columns.is_unique:\n        raise ValueError(f'{csv}: duplicate column names')\n    idcol = next((c for c in ('image', 'id_code', 'image_id', 'filename', 'file', 'id')\n                  if c in labels.columns), None)\n    gradecol = next((c for c in ('level', 'diagnosis', 'grade', 'label', 'dr_grade')\n                     if c in labels.columns), None)\n    if idcol is None or (require_grades and gradecol is None):\n        raise ValueError(f'{csv}: expected an image ID and DR grade (0-4); found {list(labels.columns)}')\n    if labels.empty or labels[idcol].isna().any():\n        raise ValueError(f'{csv}: empty labels or missing image IDs')\n    ids = labels[idcol].str.strip().str.replace('\\\\', '/', regex=False).str.rsplit('/', n=1).str[-1]\n    ids = ids.str.replace(r'(?i)\\.(jpg|jpeg|png)$', '', regex=True)\n    if ids.eq('').any() or ids.duplicated().any():\n        raise ValueError(f'{csv}: empty or duplicate image IDs after filename normalization')\n    result = pd.DataFrame({'image': ids})\n    if gradecol is not None:\n        grades = pd.to_numeric(labels[gradecol], errors='coerce')\n        if grades.isna().any() or not grades.isin(range(5)).all():\n            raise ValueError(f'{csv}: DR grades must be integers in 0-4')\n        result['level'] = grades.astype(int)\n    return result\n\n\ndef eye_labels():\n    return original_eye_labels()\n\n\ndef check_eye_splits(eye):\n    active = eye[eye.split != 'quarantine']\n    assert active.groupby('group_id').split.nunique().max() == 1, 'Patient/duplicate leakage across splits'\n    assert eye.loc[eye.source_split == 'original_test', 'split'].eq('internal_test').all()\n    assert not eye.loc[eye.source_split == 'original_train', 'split'].eq('internal_test').any()\n    assert set(active.split) == {'backbone_train', 'checkpoint_selection', 'router_train',\n                                'policy_validation', 'internal_test'}, 'An EyePACS split is empty'\n\n\ndef split_eyepacs(eye):\n    eye = eye.copy()\n    is_test = eye.source_split == 'original_test'\n    test_groups = set(eye.loc[is_test, 'group_id'])\n    eligible = eye[(~is_test) & (~eye.group_id.isin(test_groups))].copy()\n    # Preserve the original test partition and remove overlapping development groups.\n    eye['split'] = 'quarantine'\n    eye.loc[is_test, 'split'] = 'internal_test'\n    if eligible.group_id.nunique() < 18:\n        raise ValueError('Need at least 18 independent training groups after test-overlap quarantine.')\n    folds = np.full(len(eligible), -1)\n    splitter = StratifiedGroupKFold(n_splits=18, shuffle=True, random_state=CFG['seed'])\n    for f, (_, idx) in enumerate(splitter.split(eligible, eligible.grade, eligible.group_id)):\n        folds[idx] = f\n    assert (folds >= 0).all(), 'Unassigned training fold'\n    eye.loc[eligible.index, 'split'] = np.select([folds < 13, folds < 15, folds < 17],\n        ['backbone_train', 'checkpoint_selection', 'router_train'], default='policy_validation')\n    check_eye_splits(eye)\n    return eye\n\n\ndef scan(csv, folder, dataset):\n    return preprocessing_runtime().scan(csv, folder, dataset, CFG, OUT)\n\ndef prepare():\n    if (OUT/'manifest.csv').exists(): return manifest()\n    eye=scan(eye_labels(),CFG['eye_images'],'eyepacs')\n    apt=scan(CFG['aptos_csv'],CFG['aptos_images'],'aptos')\n    frame=pd.concat([eye,apt],ignore_index=True); n=len(eye); uf=UnionFind(n)\n    for _,indices in eye.groupby('patient_id').groups.items():\n        indices=list(indices)\n        for i in indices[1:]: uf.union(indices[0],i)\n    tree=BKTree(); exact={}; pairs=[]; quarantine=set()\n    for i,row in frame.iterrows():\n        h=int(row.phash_hex,16); candidates=set(tree.query(h,CFG['near_distance']))\n        candidates.update(exact.get(row.exact,[]))\n        for j in candidates:\n            if i<n: uf.union(i,j)\n            else: quarantine.add(i)\n            pairs.append(dict(a=j,b=i,distance=(h^int(frame.iloc[j].phash_hex,16)).bit_count(),\n                              exact=row.exact==frame.iloc[j].exact))\n        if i<n: tree.add(h,i); exact.setdefault(row.exact,[]).append(i)\n    pd.DataFrame(pairs,columns=['a','b','distance','exact']).to_csv(OUT/'duplicate_pairs.csv',index=False)\n    contact_sheet(frame,[(p['a'],p['b']) for p in pairs], 'duplicate_contact_sheet.png')\n    eye['group_id']=['eye:'+str(uf.find(i)) for i in range(n)]\n    eye=split_eyepacs(eye)\n    eye[['image_id','path','grade','patient_id','group_id','source_split','split']].to_csv(\n        OUT/'eyepacs_split_membership.csv',index=False)\n    apt['group_id']='apt:'+apt.image_id.astype(str)\n    apt['split']=['quarantine' if i+n in quarantine else 'external_test' for i in range(len(apt))]\n    # APTOS is evaluated and bootstrapped at image level; patient grouping is unavailable.\n    if CFG['smoke']:\n        rng=np.random.default_rng(CFG['seed']); chunks=[]\n        for _,part in eye.groupby('split'):\n            groups=part.group_id.unique(); chosen=rng.choice(groups,min(len(groups),CFG['smoke_groups']),replace=False)\n            chunks.append(part[part.group_id.isin(chosen)])\n        eye=pd.concat(chunks); apt=apt.groupby('split',group_keys=False).head(CFG['smoke_groups'])\n    prepared=pd.concat([eye,apt],ignore_index=True)\n    check_eye_splits(eye)\n    prepared.groupby(['dataset','split','grade']).size().rename('count').to_csv(OUT/'split_counts.csv')\n    prepared[prepared.insufficient_detail].to_csv(OUT/'insufficient_detail.csv',index=False)\n    conflicts=eye.groupby('exact').grade.nunique(); conflicts[conflicts>1].to_csv(OUT/'duplicate_label_conflicts.csv')\n    temporary = OUT/'manifest.csv.tmp'\n    prepared.to_csv(temporary,index=False); os.replace(temporary,OUT/'manifest.csv')\n    print(prepared.groupby(['dataset','split']).size())\n    return prepared\n\ndef manifest():\n    frame=resolve_cache_paths(pd.read_csv(require('manifest.csv'))); eye=frame[frame.dataset=='eyepacs']\n    check_eye_splits(eye)\n    assert set(eye[eye.split=='backbone_train'].group_id).isdisjoint(eye[eye.split=='router_train'].group_id)\n    return frame\n\n","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Matched backbones and resumable training\nResume is exact at epoch boundaries, not in the middle of an interrupted epoch.\nMemory probing includes optimizer allocation; accumulation weights the final\npartial effective batch correctly. A later OOM retries the epoch from disk.\n","metadata":{}},{"cell_type":"code","source":"def transform(size,training=False):\n    ops=[transforms.Resize((size,size))]\n    if training: ops += [transforms.RandomHorizontalFlip(),transforms.RandomVerticalFlip(),\n                        transforms.RandomRotation(15),transforms.ColorJitter(.1,.1,.1)]\n    return transforms.Compose(ops+[transforms.ToTensor(),transforms.Normalize([.485,.456,.406],[.229,.224,.225])])\n\nclass Fundus(Dataset):\n    def __init__(self,frame,size,training=False): self.frame=frame.reset_index(drop=True); self.tf=transform(size,training)\n    def __len__(self): return len(self.frame)\n    def __getitem__(self,i):\n        row=self.frame.iloc[i]\n        rgb=load_fundus(row.path)\n        return self.tf(Image.fromarray(rgb)), int(row.grade), i\n\ndef make_model():\n    model=models.efficientnet_b0(weights=None)\n    model.classifier[1]=nn.Linear(model.classifier[1].in_features,5)\n    return model\n\ndef ensure_initial():\n    path=OUT/'initial.pt'\n    if path.exists(): return path\n    seed_all(CFG['seed'])\n    base=models.efficientnet_b0(weights=None)\n    if CFG['pretrained_weights']:\n        weights=torch.load(CFG['pretrained_weights'],map_location='cpu',weights_only=True)\n        base.load_state_dict(weights)\n    elif CFG['allow_download']:\n        base=models.efficientnet_b0(weights=models.EfficientNet_B0_Weights.DEFAULT)\n    else:\n        cached=Path(torch.hub.get_dir())/'checkpoints'/Path(models.EfficientNet_B0_Weights.DEFAULT.url).name\n        if not cached.exists():\n            raise RuntimeError('Attach torchvision EfficientNet-B0 ImageNet state_dict and set pretrained_weights, or enable internet and allow_download.')\n        base.load_state_dict(torch.load(cached,map_location='cpu',weights_only=True))\n    base.classifier[1]=nn.Linear(base.classifier[1].in_features,5)\n    atomic_torch(base.state_dict(),path); return path\n\ndef rng_state():\n    return dict(python=random.getstate(),numpy=np.random.get_state(),torch=torch.get_rng_state(),\n                cuda=torch.cuda.get_rng_state_all() if DEVICE.type=='cuda' else None)\ndef restore_rng(state):\n    random.setstate(state['python']); np.random.set_state(state['numpy']); torch.set_rng_state(state['torch'])\n    if DEVICE.type=='cuda' and state['cuda'] is not None: torch.cuda.set_rng_state_all(state['cuda'])\n\ndef load_checkpoint(path):\n    # Only load trusted outputs created by this notebook; resume contains Python RNG objects.\n    data=torch.load(path,map_location='cpu',weights_only=False)\n    assert data['signature']==signature\n    return data\n\ndef scaler(): return torch.amp.GradScaler('cuda',enabled=DEVICE.type=='cuda')\ndef autocast(): return torch.autocast(device_type=DEVICE.type,enabled=DEVICE.type=='cuda')\n\ndef probe(size):\n    for batch in (16,8,4,2,1):\n        if batch>min(CFG['effective_batch'], CFG['max_microbatch']): continue\n        model=None; opt=None; x=None; loss=None\n        try:\n            model=make_model().to(DEVICE); opt=torch.optim.AdamW(model.parameters())\n            x=torch.randn(batch,3,size,size,device=DEVICE)\n            with autocast(): loss=model(x).sum()\n            loss.backward(); opt.step(); return batch\n        except torch.cuda.OutOfMemoryError:\n            if batch==1: raise\n        finally:\n            del model,opt,x,loss; gc.collect()\n            if DEVICE.type=='cuda': torch.cuda.empty_cache()\n\ndef qwk(y,p):\n    value=cohen_kappa_score(y,p,labels=list(range(5)),weights='quadratic')\n    return float(value) if np.isfinite(value) else np.nan\n\n@torch.inference_mode()\ndef validation(model,frame,size,batch):\n    model.eval(); preds=[]\n    for x,_,_ in DataLoader(Fundus(frame,size),batch_size=batch,num_workers=CFG['workers']):\n        with autocast(): preds.extend(model(x.to(DEVICE)).argmax(1).cpu().tolist())\n    return qwk(frame.grade.to_numpy(),preds)\n\ndef train(size):\n    initial=ensure_initial(); frame=manifest()\n    training=frame[frame.split=='backbone_train']; valid=frame[frame.split=='checkpoint_selection']\n    latest=OUT/f'latest_{size}.pt'; bestpath=OUT/f'best_{size}.pt'\n    if latest.exists() and load_checkpoint(latest)['epoch']>=CFG['epochs']:\n        require(f'best_{size}.pt'); print(f'{size}: completed'); return\n    micro=probe(size); seed_all(CFG['seed'])\n    model=make_model().to(DEVICE); model.load_state_dict(torch.load(initial,weights_only=True,map_location='cpu'))\n    opt=torch.optim.AdamW(model.parameters(),lr=CFG['lr'],weight_decay=CFG['weight_decay'])\n    sched=torch.optim.lr_scheduler.CosineAnnealingLR(opt,T_max=CFG['epochs']); amp=scaler()\n    def save(epoch,best):\n        atomic_torch(dict(signature=signature,model=model.state_dict(),optimizer=opt.state_dict(),\n            scheduler=sched.state_dict(),scaler=amp.state_dict(),epoch=epoch,best=best,\n            rng=rng_state(),config=CFG,microbatch=micro),latest)\n    if not latest.exists(): save(0,-float('inf'))\n    while True:\n        state=load_checkpoint(latest); model.load_state_dict(state['model']); opt.load_state_dict(state['optimizer'])\n        sched.load_state_dict(state['scheduler']); amp.load_state_dict(state['scaler']); restore_rng(state['rng'])\n        start,best=state['epoch'],state['best']\n        try:\n            for epoch in range(start,CFG['epochs']):\n                model.train(); loader=DataLoader(Fundus(training,size,True),batch_size=micro,shuffle=True,\n                    num_workers=CFG['workers'],pin_memory=DEVICE.type=='cuda')\n                opt.zero_grad(set_to_none=True); seen=0; window=0; total=0.\n                for x,y,_ in loader:\n                    if window==0: target=min(CFG['effective_batch'],len(training)-seen)\n                    with autocast(): loss=nn.functional.cross_entropy(model(x.to(DEVICE)),y.to(DEVICE),reduction='sum')/target\n                    amp.scale(loss).backward(); window+=len(y); seen+=len(y); total+=float(loss.detach())*target\n                    if window==target:\n                        amp.step(opt); amp.update(); opt.zero_grad(set_to_none=True); window=0\n                score=validation(model,valid,size,micro); sched.step()\n                if np.isfinite(score) and score>best:\n                    best=score; atomic_torch(dict(signature=signature,model=model.state_dict(),epoch=epoch+1,qwk=score),bestpath)\n                if not bestpath.exists(): raise RuntimeError('Checkpoint QWK undefined; increase smoke sample or inspect labels.')\n                save(epoch+1,best)\n                print(f'{size} epoch {epoch+1}: loss={total/len(training):.4f}, QWK={score:.4f}')\n            break\n        except torch.cuda.OutOfMemoryError:\n            opt.zero_grad(set_to_none=True); gc.collect(); torch.cuda.empty_cache()\n            if micro==1: raise\n            micro//=2; print('Retrying saved epoch with microbatch',micro)\n    del model,opt; gc.collect()\n    if DEVICE.type=='cuda': torch.cuda.empty_cache()\n\ndef backbone(size):\n    model=make_model(); model.load_state_dict(load_checkpoint(require(f'best_{size}.pt'))['model'])\n    return model.to(DEVICE).eval()\n\n","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Cached development predictions and benefit regression\nNo training-time labels or high-resolution features enter deployed router inputs.\n","metadata":{}},{"cell_type":"code","source":"@torch.inference_mode()\ndef features(model,x):\n    z=model.avgpool(model.features(x)).flatten(1)\n    return model.classifier(z).softmax(1),z\n\ndef collect(split):\n    path=OUT/f'pred_{split}.npz'\n    if path.exists(): return path\n    frame=manifest(); frame=frame[frame.split==split].reset_index(drop=True)\n    if frame.empty: raise RuntimeError(f'No eligible images in {split}')\n    probs={}; embedding=None\n    for size in (224,448):\n        model=backbone(size); pp=[]; zz=[]\n        with torch.inference_mode():\n            for x,_,_ in DataLoader(Fundus(frame,size),batch_size=1,num_workers=CFG['workers']):\n                with autocast(): p,z=features(model,x.to(DEVICE))\n                pp.append(p.float().cpu().numpy())\n                if size==224: zz.append(z.float().cpu().numpy())\n        probs[size]=np.concatenate(pp)\n        if size==224: embedding=np.concatenate(zz)\n        del model; gc.collect()\n        if DEVICE.type=='cuda': torch.cuda.empty_cache()\n    qq=[]\n    for p in frame.path:\n        rgb=load_fundus(p)\n        qq.append(quality(cv2.resize(rgb,(224,224))))\n    tmp=Path(str(path)+'.tmp')\n    with tmp.open('wb') as f:\n        np.savez_compressed(f,ids=frame.image_id.to_numpy(dtype=str),groups=frame.group_id.to_numpy(dtype=str),\n            paths=frame.path.to_numpy(dtype=str),y=frame.grade.to_numpy(),p224=probs[224],p448=probs[448],\n            embeddings=embedding,quality=np.asarray(qq,dtype=np.float32))\n    os.replace(tmp,path); return path\n\ndef read_predictions(split):\n    with np.load(require(f'pred_{split}.npz'),allow_pickle=False) as f: return {k:f[k] for k in f.files}\n\ndef entropy(p): return -(p*np.log(np.clip(p,1e-8,1))).sum(1)\ndef router_inputs(data,embeddings=True):\n    p=data['p224']; ordered=np.sort(p,axis=1)\n    chunks=[p,entropy(p)[:,None],(ordered[:,-1]-ordered[:,-2])[:,None],data['quality']]\n    if embeddings: chunks.append(data['embeddings'])\n    return np.concatenate(chunks,axis=1)\ndef benefit(data):\n    return (data['y']-data['p224'].argmax(1))**2-(data['y']-data['p448'].argmax(1))**2\ndef threshold(scores,budget):\n    n=len(scores); k=int(np.floor(n*budget))\n    if k==0: return float(np.max(scores))\n    if k==n: return float('-inf')\n    order=np.sort(scores); return float(order[-k-1])  # strict >, ties conservatively reduce routing\n\ndef fit_router():\n    if (OUT/'policies.json').exists():\n        require('router.joblib'); require('router_no_embeddings.joblib'); return\n    data=read_predictions('router_train'); valid=read_predictions('policy_validation')\n    policies=[]\n    for name,use in [('learned',True),('no_embeddings',False)]:\n        model=make_pipeline(StandardScaler(),Ridge(alpha=10))\n        model.fit(router_inputs(data,use),benefit(data))\n        router_path=OUT/('router.joblib' if use else 'router_no_embeddings.joblib')\n        joblib.dump(model,router_path)\n        reloaded=joblib.load(router_path)\n        assert np.allclose(model.predict(router_inputs(valid,use)),reloaded.predict(router_inputs(valid,use)))\n        scores=model.predict(router_inputs(valid,use))\n        for budget in CFG['budgets']:\n            policies.append(dict(name=f'{name}_{budget:g}',kind=name,budget=budget,threshold=threshold(scores,budget)))\n    for budget in CFG['budgets']:\n        policies.append(dict(name=f'entropy_{budget:g}',kind='entropy',budget=budget,\n                             threshold=threshold(entropy(valid['p224']),budget)))\n    (OUT/'policies.json').write_text(json.dumps(dict(selected_on='policy_validation',policies=policies),indent=2))\n\ndef policies():\n    data=json.loads(require('policies.json').read_text()); assert data['selected_on']=='policy_validation'\n    return data['policies']\ndef mask_for(data,policy):\n    kind=policy['kind']\n    if kind=='always224': return np.zeros(len(data['y']),bool)\n    if kind=='always448': return np.ones(len(data['y']),bool)\n    if kind=='entropy': scores=entropy(data['p224'])\n    else:\n        use=kind=='learned'; router=joblib.load(require('router.joblib' if use else 'router_no_embeddings.joblib'))\n        scores=router.predict(router_inputs(data,use))\n    return scores>policy['threshold']\ndef routed_predictions(data,mask): return np.where(mask,data['p448'].argmax(1),data['p224'].argmax(1))\n\n@torch.inference_mode()\ndef predict_image(path, low, high, policy=None, router=None):\n    \"\"\"Deployment interface: no labels or cached high-resolution predictions.\"\"\"\n    if policy is None: policy=json.loads(require('selected_policy.json').read_text())['policy']\n    rgb=load_fundus(path)\n    image=Image.fromarray(rgb)\n    if policy['kind']=='always448':\n        with autocast(): p=high(transform(448)(image).unsqueeze(0).to(DEVICE)).softmax(1)\n        return dict(grade=int(p.argmax(1).item()),routed=True,probabilities=p.float().cpu().numpy()[0])\n    with autocast(): p,z=features(low,transform(224)(image).unsqueeze(0).to(DEVICE))\n    cpu=p.float().cpu().numpy(); go=False\n    if policy['kind']=='entropy': go=bool(entropy(cpu)[0]>policy['threshold'])\n    elif policy['kind'] in ('learned','no_embeddings'):\n        use=policy['kind']=='learned'\n        if router is None: router=joblib.load(require('router.joblib' if use else 'router_no_embeddings.joblib'))\n        d=dict(p224=cpu,embeddings=z.float().cpu().numpy(),quality=np.asarray(quality(cv2.resize(rgb,(224,224))))[None])\n        go=bool(router.predict(router_inputs(d,use))[0]>policy['threshold'])\n    if go:\n        with autocast(): p=high(transform(448)(image).unsqueeze(0).to(DEVICE)).softmax(1)\n    return dict(grade=int(p.argmax(1).item()),routed=go,probabilities=p.float().cpu().numpy()[0])\n\n","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Actual cascade timing and validation-only operating policy selection\nBenchmark before final evaluation. One GPU, batch one, both models resident.\nModel-only includes device tensors and routing; end-to-end also includes file\ndecode, original-crop preprocessing, resizing, and transfer. Cached predictions\nare never substituted for model execution during timing.\n","metadata":{}},{"cell_type":"code","source":"def sync():\n    if DEVICE.type=='cuda': torch.cuda.synchronize()\ndef latency_summary(values):\n    return dict(mean_ms=float(np.mean(values)),median_ms=float(np.median(values)),p95_ms=float(np.percentile(values,95)))\n\ndef benchmark_split(split, policy_list):\n    frame=manifest(); frame=frame[frame.split==split].reset_index(drop=True)\n    frame=frame.sample(min(len(frame),CFG['benchmark_images']),random_state=CFG['seed'])\n    low,high=backbone(224),backbone(448)\n    routers={k:joblib.load(require(f)) for k,f in [('learned','router.joblib'),('no_embeddings','router_no_embeddings.joblib')]}\n    t224,t448=transform(224),transform(448); records=[]\n    @torch.inference_mode()\n    def execute(x1,x2,q,policy):\n        with autocast():\n            if policy['kind']=='always448': return high(x2).argmax(1),True\n            p,z=features(low,x1)\n        kind=policy['kind']\n        if kind=='always224': return p.argmax(1),False\n        cpu=p.float().cpu().numpy()\n        if kind=='entropy': score=entropy(cpu)[0]\n        else:\n            d=dict(p224=cpu,embeddings=z.float().cpu().numpy(),quality=q[None])\n            score=routers[kind].predict(router_inputs(d,kind=='learned'))[0]\n        go=score>policy['threshold']\n        if go:\n            with autocast(): return high(x2).argmax(1),True\n        return p.argmax(1),False\n    for policy in policy_list:\n        model_times=[]; full_times=[]; flags=[]\n        for count,row in enumerate(frame.itertuples()):\n            rgb=load_fundus(row.path)\n            image=Image.fromarray(rgb); q=np.asarray(quality(cv2.resize(rgb,(224,224))))\n            x1=t224(image).unsqueeze(0).to(DEVICE); x2=t448(image).unsqueeze(0).to(DEVICE)\n            if count==0:\n                for _ in range(CFG['warmup']): execute(x1,x2,q,policy)\n            sync(); start=time.perf_counter(); _,go=execute(x1,x2,q,policy); sync()\n            model_times.append((time.perf_counter()-start)*1000)\n            sync(); start=time.perf_counter()\n            fresh=preprocessing_runtime().model_rgb(row.source_path, CFG)\n            fresh_image=Image.fromarray(fresh)\n            # Low route decision first; prepare high input only if required.\n            if policy['kind']=='always448':\n                _,go=execute(None,t448(fresh_image).unsqueeze(0).to(DEVICE),None,policy)\n            else:\n                fresh_x=t224(fresh_image).unsqueeze(0).to(DEVICE)\n                if policy['kind']=='always224': _,go=execute(fresh_x,None,None,policy)\n                else:\n                    with torch.inference_mode(),autocast(): p,z=features(low,fresh_x)\n                    cpu=p.float().cpu().numpy()\n                    fresh_q=np.asarray(quality(cv2.resize(fresh,(224,224))))\n                    if policy['kind']=='entropy': score=entropy(cpu)[0]\n                    else:\n                        d=dict(p224=cpu,embeddings=z.float().cpu().numpy(),quality=fresh_q[None])\n                        score=routers[policy['kind']].predict(router_inputs(d,policy['kind']=='learned'))[0]\n                    go=score>policy['threshold']\n                    if go:\n                        with torch.inference_mode(),autocast(): high(t448(fresh_image).unsqueeze(0).to(DEVICE))\n            sync(); full_times.append((time.perf_counter()-start)*1000); flags.append(go)\n        for mode,values in [('model',model_times),('end_to_end',full_times)]:\n            records.append(dict(split=split,policy=policy['name'],mode=mode,route_rate=np.mean(flags),**latency_summary(values)))\n    del low,high; gc.collect()\n    if DEVICE.type=='cuda': torch.cuda.empty_cache()\n    return pd.DataFrame(records)\n\nBASE=[dict(name='always224',kind='always224'),dict(name='always448',kind='always448')]\ndef benchmark():\n    if (OUT/'selected_policy.json').exists(): return\n    timing=benchmark_split('policy_validation',BASE+policies()); timing.to_csv(OUT/'latency_validation.csv',index=False)\n    data=read_predictions('policy_validation'); ref=routed_predictions(data,np.ones(len(data['y']),bool))\n    reference=metrics(data['y'],ref); candidates=[]\n    for p in policies():\n        if p['kind']!='learned': continue\n        score=metrics(data['y'],routed_predictions(data,mask_for(data,p)))\n        cost=float(timing[(timing.policy==p['name'])&(timing['mode']=='end_to_end')].mean_ms.iloc[0])\n        qualifies=score['qwk']>=reference['qwk']-.01 and score['serious_undergrading']<=reference['serious_undergrading']+.01\n        candidates.append(dict(**p,**score,cost_ms=cost,qualifies=bool(qualifies)))\n    eligible=[p for p in candidates if p['qualifies'] and p['cost_ms']<float(timing[(timing.policy=='always448')&(timing['mode']=='end_to_end')].mean_ms.iloc[0])]\n    selected=min(eligible,key=lambda p:(p['cost_ms'],p['name'])) if eligible else BASE[1]\n    (OUT/'selected_policy.json').write_text(json.dumps(dict(selected_on='policy_validation',policy=selected,\n        candidates=candidates,fallback=not bool(eligible),criteria='QWK within .01; undergrading increase <= .01; faster than always448'),indent=2))\n\n","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Frozen internal and external evaluation\nPaired bootstrap differences use always448 as reference. EyePACS resamples\ncomplete patient/duplicate groups; APTOS resamples images. Undefined statistics\nare recorded as NaN and effective bootstrap replicate counts are exported.\nRandom baselines match exact routing counts; oracle uses labels and is diagnostic.\n\nEyePACS `internal_test` is the original test CSV partition; its labels are never\nused for backbone training, checkpoint selection, router fitting, or policy selection.\n","metadata":{}},{"cell_type":"code","source":"def metrics(y,p):\n    cm=confusion_matrix(y,p,labels=range(5)); recall=np.divide(np.diag(cm),cm.sum(1),out=np.full(5,np.nan),where=cm.sum(1)>0)\n    pos=y>=2; negative=~pos\n    return dict(qwk=qwk(y,p),accuracy=float(np.mean(y==p)),mae=float(np.abs(y-p).mean()),\n        serious_undergrading=float(np.mean(y-p>=2)),\n        referable_sensitivity=float(np.mean(p[pos]>=2)) if pos.any() else np.nan,\n        referable_specificity=float(np.mean(p[negative]<2)) if negative.any() else np.nan,\n        **{f'recall_grade_{i}':float(v) for i,v in enumerate(recall)})\n\ndef bootstrap_difference(data,p,reference,grouped):\n    y=data['y']; groups=data['groups'] if grouped else np.arange(len(y)).astype(str)\n    unique,inverse=np.unique(groups,return_inverse=True); members=[np.flatnonzero(inverse==i) for i in range(len(unique))]\n    rng=np.random.default_rng(CFG['seed']); keys=['qwk','serious_undergrading','mae','referable_sensitivity']; draws={k:[] for k in keys}\n    for _ in range(CFG['bootstrap']):\n        idx=np.concatenate([members[i] for i in rng.integers(0,len(unique),len(unique))])\n        a,b=metrics(y[idx],p[idx]),metrics(y[idx],reference[idx])\n        for k in keys: draws[k].append(a[k]-b[k])\n    result={}\n    for k,values in draws.items():\n        valid=np.asarray(values); valid=valid[np.isfinite(valid)]\n        bounds=np.percentile(valid,[2.5,97.5]) if len(valid) else [np.nan,np.nan]\n        result.update({k+'_difference':metrics(y,p)[k]-metrics(y,reference)[k],k+'_ci_low':bounds[0],k+'_ci_high':bounds[1],k+'_bootstrap_n':len(valid)})\n    return result\n\ndef evaluate():\n    selected=json.loads(require('selected_policy.json').read_text()); assert selected['selected_on']=='policy_validation'\n    all_rows=[]; intervals=[]; latency=[]\n    for split in ('internal_test','external_test'):\n        collect(split); data=read_predictions(split); n=len(data['y']); delta=benefit(data)\n        rows=[]; rng=np.random.default_rng(CFG['seed']); order=rng.permutation(n)\n        candidates=[(p,mask_for(data,p)) for p in BASE+policies()]\n        for policy,mask in list(candidates):\n            if policy['kind'].startswith('always'): continue\n            k=int(mask.sum()); random_mask=np.zeros(n,bool); random_mask[order[:k]]=True\n            oracle=np.zeros(n,bool); oracle[np.argsort(-delta,kind='stable')[:k]]=True\n            candidates += [(dict(name='random_matched_'+policy['name'],kind='random'),random_mask),\n                           (dict(name='oracle_diagnostic_'+policy['name'],kind='oracle'),oracle)]\n        output=pd.DataFrame(dict(image_id=data['ids'],group_id=data['groups'],reference_grade=data['y'],\n            pred224=data['p224'].argmax(1),pred448=data['p448'].argmax(1),benefit=delta,\n            low_confidence=data['p224'].max(1)))\n        for i,key in enumerate(['blur','brightness','contrast','clipped_fraction']): output[key]=data['quality'][:,i]\n        ref=data['p448'].argmax(1)\n        for policy,mask in candidates:\n            pred=routed_predictions(data,mask)\n            assert np.array_equal(routed_predictions(data,np.zeros(n,bool)),data['p224'].argmax(1))\n            assert np.array_equal(routed_predictions(data,np.ones(n,bool)),ref)\n            missed=(delta>0)&(~mask)&(data['p224'].max(1)>=.9)\n            rows.append(dict(split=split,policy=policy['name'],diagnostic=policy['kind']=='oracle',\n                selected=policy['name']==selected['policy']['name'],route_rate=float(mask.mean()),\n                improved=int(((delta>0)&mask).sum()),harmed=int(((delta<0)&mask).sum()),\n                beneficial_recall=float(mask[delta>0].mean()) if (delta>0).any() else np.nan,\n                confidently_wrong_missed=int(missed.sum()),**metrics(data['y'],pred)))\n            intervals.append(dict(split=split,policy=policy['name'],**bootstrap_difference(data,pred,ref,split=='internal_test')))\n            output[policy['name']+'_prediction']=pred; output[policy['name']+'_routed']=mask\n            pd.DataFrame(confusion_matrix(data['y'],pred,labels=range(5))).to_csv(OUT/f'confusion_{split}_{policy[\"name\"]}.csv',index=False)\n            if policy['name']==selected['policy']['name']:\n                output.loc[missed].to_csv(OUT/f'confident_misses_{split}.csv',index=False)\n                analysis=output.assign(beneficial=delta>0,escalated=mask,confident_missed=missed,\n                    confidence_bin=pd.cut(output.low_confidence,[0,.5,.7,.9,1],include_lowest=True))\n                analysis.groupby(['reference_grade','confidence_bin'],observed=True).agg(\n                    count=('image_id','size'),beneficial=('beneficial','sum'),\n                    routed=('escalated','sum'),confident_missed=('confident_missed','sum'),\n                    mean_blur=('blur','mean'),mean_brightness=('brightness','mean')).to_csv(OUT/f'failure_analysis_{split}.csv')\n                examples=[]\n                for kind in ((delta>0)&mask,(delta<0)&mask,missed):\n                    ids=np.flatnonzero(kind)[:8]; examples.extend((int(i),int(i)) for i in ids)\n                contact_sheet(pd.DataFrame(dict(path=data['paths'],image_id=[f'{i}: y={y} low={l} high={h}' for i,y,l,h in zip(data['ids'],data['y'],data['p224'].argmax(1),ref)])),examples,f'examples_{split}.png')\n        output.to_csv(OUT/f'predictions_{split}.csv',index=False); all_rows.extend(rows)\n        latency.append(benchmark_split(split,BASE+policies()))\n        # Four exact-grading outcome counts, plus ordinal benefit/harm.\n        low=data['p224'].argmax(1)==data['y']; high=ref==data['y']\n        pd.DataFrame([dict(both_correct=int((low&high).sum()),low_wrong_high_correct=int((~low&high).sum()),\n            low_correct_high_wrong=int((low&~high).sum()),both_wrong=int((~low&~high).sum()),\n            ordinal_improvement=int((delta>0).sum()),ordinal_harm=int((delta<0).sum()))]).to_csv(OUT/f'benefit_outcomes_{split}.csv',index=False)\n    results=pd.DataFrame(all_rows); results.to_csv(OUT/'metrics.csv',index=False)\n    pd.DataFrame(intervals).to_csv(OUT/'paired_bootstrap.csv',index=False)\n    times=pd.concat(latency); times.to_csv(OUT/'latency_test.csv',index=False)\n    merged=results.merge(times[times['mode']=='end_to_end'],on=['split','policy'],suffixes=('','_timing'))\n    for split,part in merged.groupby('split'):\n        fig,axes=plt.subplots(1,2,figsize=(12,4))\n        for prefix in ('always','learned','no_embeddings','entropy'):\n            chunk=part[part.policy.str.startswith(prefix)].sort_values('mean_ms')\n            for ax,metric in zip(axes,['qwk','serious_undergrading']):\n                ax.plot(chunk.mean_ms,chunk[metric],'o-',label=prefix); ax.set_xlabel('Measured end-to-end mean latency (ms)'); ax.set_ylabel(metric)\n        axes[0].legend(); fig.tight_layout(); fig.savefig(OUT/f'cost_curve_{split}.png'); plt.close(fig)\n    print(results.to_string(index=False)); print('Selected policy:',selected['policy']['name'])\n\n","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Invariant checks and stage execution\nRun all stages only after checking paths, GPU availability, and attached weights.\nPersist /kaggle/working outputs as a Kaggle dataset between sessions. Restore\nthe matching full/ or smoke/ directory, with unchanged experiment configuration.\nEvaluation is intentionally repeatable; it never changes saved routing thresholds.\n\nAfter switching from original to resized EyePACS, use a fresh output directory.\nPrior manifests/checkpoints have a different experiment signature and cannot be reused.\n","metadata":{}},{"cell_type":"code","source":"def self_checks():\n    d=dict(y=np.array([3,3,3]),p224=np.eye(5)[[1,3,2]],p448=np.eye(5)[[3,2,2]])\n    assert np.array_equal(benefit(d),[4,-1,0])\n    assert np.array_equal(routed_predictions(d,np.zeros(3,bool)),[1,3,2])\n    assert np.array_equal(routed_predictions(d,np.ones(3,bool)),[3,2,2])\n    assert (np.array([1.,2.,3.,4.])>threshold(np.array([1.,2.,3.,4.]),.5)).sum()==2\n    tree=BKTree(); tree.add(0,0); tree.add(3,1); assert set(tree.query(0,1))=={0}\n    # Small real torch continuation test: checkpoint and RNG recovery must\n    # reproduce the next optimizer step, including momentum state.\n    original_rng=rng_state()\n    with tempfile.TemporaryDirectory(dir=OUT) as tmp:\n        a=nn.Linear(2,1); optimizer=torch.optim.AdamW(a.parameters(),lr=.01)\n        def step(net,opt):\n            x=torch.randn(4,2); opt.zero_grad(); loss=net(x).square().mean(); loss.backward(); opt.step()\n        step(a,optimizer)\n        path=Path(tmp)/'resume.pt'\n        atomic_torch(dict(signature=signature,model=a.state_dict(),optimizer=optimizer.state_dict(),rng=rng_state()),path)\n        step(a,optimizer); expected={k:v.clone() for k,v in a.state_dict().items()}\n        b=nn.Linear(2,1); other=torch.optim.AdamW(b.parameters(),lr=.01)\n        checkpoint=load_checkpoint(path); b.load_state_dict(checkpoint['model']); other.load_state_dict(checkpoint['optimizer'])\n        restore_rng(checkpoint['rng']); step(b,other)\n        assert all(torch.equal(expected[k],v) for k,v in b.state_dict().items()), 'Checkpoint continuation mismatch'\n    restore_rng(original_rng)\n    print('Core invariants passed. Mode:', 'SMOKE / DEBUG ONLY' if CFG['smoke'] else 'FULL')\n\nself_checks()\nSTAGES=dict(prepare=prepare,train224=lambda:train(224),train448=lambda:train(448),\n    predict_dev=lambda:[collect(s) for s in ('router_train','policy_validation')],\n    router=fit_router,benchmark=benchmark,evaluate=evaluate)\nassert RUN_STAGE in list(STAGES)+['all']\nfor stage in (list(STAGES) if RUN_STAGE=='all' else [RUN_STAGE]):\n    print('Running',stage); STAGES[stage]()\n","metadata":{},"outputs":[],"execution_count":null}]}