{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.11.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"nvidiaTeslaT4","dataSources":[{"sourceId":117193,"databundleVersionId":14001224,"sourceType":"competition"}],"dockerImageVersionId":31153,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import os\nimport random\nimport numpy as np\nimport pandas as pd\nfrom PIL import Image\nfrom tqdm.auto import tqdm\n\nimport torch\nimport torch.nn as nn\nfrom torch.utils.data import Dataset, DataLoader\nimport torchvision.transforms as T\n\nfrom transformers import AutoFeatureExtractor, AutoModel\n\nfrom sklearn.metrics import roc_auc_score, accuracy_score\nfrom sklearn.model_selection import train_test_split\n\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nprint(\"Device:\", device)","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2025-10-16T22:30:23.908094Z","iopub.execute_input":"2025-10-16T22:30:23.908801Z","iopub.status.idle":"2025-10-16T22:30:36.938060Z","shell.execute_reply.started":"2025-10-16T22:30:23.908774Z","shell.execute_reply":"2025-10-16T22:30:36.937293Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# from huggingface_hub import snapshot_download\n# from insightface.app import FaceAnalysis\n# import numpy as np\n# import cv2\n\n# snapshot_download(\n#     \"fal/AuraFace-v1\",\n#     local_dir=\"models/auraface\",\n# )\n# face_app = FaceAnalysis(\n#     name=\"auraface\",\n#     providers=[\"CUDAExecutionProvider\", \"CPUExecutionProvider\"],\n#     root=\".\",\n# )\n\n# input_image = cv2.imread(\"test.png\")\n\n# cv2_image = np.array(input_image.convert(\"RGB\"))\n\n# cv2_image = cv2_image[:, :, ::-1]\n# faces = face_app.get(cv2_image)\n# embedding = faces[0].normed_embedding\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-16T22:30:36.939763Z","iopub.execute_input":"2025-10-16T22:30:36.940520Z","iopub.status.idle":"2025-10-16T22:30:36.943945Z","shell.execute_reply.started":"2025-10-16T22:30:36.940489Z","shell.execute_reply":"2025-10-16T22:30:36.943207Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"DATA_DIR = \"/kaggle/input/anip-reconnaissance-faciale-estimation-ages-ocr/dataset_tache_1/dataset_tache_1\"\nTRAIN_IMG_DIR = os.path.join(DATA_DIR, \"train\")\nTEST_IMG_DIR  = os.path.join(DATA_DIR, \"test\")\n\nassert os.path.isdir(TRAIN_IMG_DIR), f\"Train images not found: {TRAIN_IMG_DIR}\"\nassert os.path.isdir(TEST_IMG_DIR), f\"Test images not found: {TEST_IMG_DIR}\"","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-16T22:30:36.944753Z","iopub.execute_input":"2025-10-16T22:30:36.944942Z","iopub.status.idle":"2025-10-16T22:30:36.965512Z","shell.execute_reply.started":"2025-10-16T22:30:36.944927Z","shell.execute_reply":"2025-10-16T22:30:36.964767Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport random\nimport cv2\nimport numpy as np\nimport torch\nfrom transformers import SegformerImageProcessor, SegformerForSemanticSegmentation\nfrom albumentations import Compose, OneOf, Rotate, Affine, Perspective, RandomBrightnessContrast, HueSaturationValue, GaussianBlur, MotionBlur\n\n# — Configuration — #\n\nIMG_DIR = \"/kaggle/input/anip-reconnaissance-faciale-estimation-ages-ocr/dataset_tache_1/dataset_tache_1/train\"\nAUG_DIR = \"/kaggle/working/aug_images\"\n\nAUGMENT_RATE = 0.45\nMIN_VARIANTS = 1\nMAX_VARIANTS = 4\nMASK_PROB = 0.5\nMIN_TRANSFORMS = 1\nMAX_TRANSFORMS = 3\nMASK_OPTIONS = [\n    \"remove_top_third\", \"remove_bottom_third\",\n    \"remove_left_half\", \"remove_right_half\",\n    \"remove_eyes\", \"remove_chin\", \"remove_forehead\"\n]\nMASK_MODES = [\"black\", \"blur\", \"mix\"]\n\nTARGET_SIZE = (256, 256)  # width, height\n\n# --- segmentation model setup ---\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nprocessor = SegformerImageProcessor.from_pretrained(\"jonathandinu/face-parsing\")\nmodel = SegformerForSemanticSegmentation.from_pretrained(\"jonathandinu/face-parsing\")\nmodel.to(device)\nmodel.eval()\n\nLABEL_MAP = {\n    \"skin\": [1],\n    \"nose\": [2],\n    \"eye_g\": [3],\n    \"left_eye\": [4],\n    \"right_eye\": [5],\n    \"l_brow\": [6],\n    \"r_brow\": [7],\n    \"mouth\": [10],\n    \"u_lip\": [11],\n    \"l_lip\": [12],\n    \"hair\": [13],\n    \"hat\": [14],\n    \"ear_l\": [8],\n    \"ear_r\": [9],\n    \"neck_l\": [16],\n    \"neck\": [17],\n    \"cloth\": [18],\n}\n\ndef get_label_map(image: np.ndarray) -> np.ndarray:\n    inputs = processor(images=image, return_tensors=\"pt\").to(device)\n    with torch.no_grad():\n        outputs = model(**inputs)\n    logits = outputs.logits\n    up = torch.nn.functional.interpolate(\n        logits, size=image.shape[:2], mode=\"bilinear\", align_corners=False\n    )\n    label_map = torch.argmax(up[0], dim=0).cpu().numpy()\n    return label_map\n\ndef make_face_mask(label_map: np.ndarray) -> np.ndarray:\n    face_labels = []\n    for region, labels in LABEL_MAP.items():\n        if region in [\"skin\", \"nose\", \"left_eye\", \"right_eye\", \"mouth\", \"u_lip\", \"l_lip\", \"l_brow\", \"r_brow\"]:\n            face_labels.extend(labels)\n    mask = np.isin(label_map, face_labels).astype(np.uint8)\n    return mask\n\ndef get_region_mask_by_option(label_map: np.ndarray, face_mask: np.ndarray, option: str) -> np.ndarray:\n    h, w = label_map.shape\n    mask_vis = face_mask.copy()\n    # (your existing logic for removing top, bottom, etc.)\n    ys, xs = np.where(face_mask == 1)\n    if option == \"remove_top_third\":\n        if len(ys) > 0:\n            y0, y1 = ys.min(), ys.max()\n            tier = (y1 - y0 + 1) / 4.0\n            boundary1 = int(y0 + tier)\n            mask_vis[y0:boundary1, :] = 0\n    elif option == \"remove_bottom_third\":\n        if len(ys) > 0:\n            y0, y1 = ys.min(), ys.max()\n            tier = (y1 - y0 + 1) / 4.0\n            boundary2 = int(y0 + 2 * tier)\n            mask_vis[boundary2 : y1 + 1, :] = 0\n    elif option == \"remove_left_half\":\n        mid = w // 3\n        mask_vis[:, :mid] = 0\n    elif option == \"remove_right_half\":\n        mid = w // 3\n        mask_vis[:, mid:] = 0\n    elif option == \"remove_eyes\":\n        eye_labels = LABEL_MAP[\"left_eye\"] + LABEL_MAP[\"right_eye\"]\n        eye_mask = np.isin(label_map, eye_labels)\n        mask_vis[eye_mask] = 0\n    elif option == \"remove_chin\":\n        if len(ys) > 0:\n            y0, y1 = ys.min(), ys.max()\n            tier = (y1 - y0 + 1) / 3.0\n            boundary2 = int(y0 + 2 * tier)\n            mask_vis[boundary2 : y1 + 1, :] = 0\n    elif option == \"remove_forehead\":\n        if len(ys) > 0:\n            y0, y1 = ys.min(), ys.max()\n            tier = (y1 - y0 + 1) / 3.0\n            boundary1 = int(y0 + tier)\n            mask_vis[y0:boundary1, :] = 0\n    return mask_vis\n\ndef apply_mask_to_image(img: np.ndarray, mask_vis: np.ndarray, mode=\"black\", blur_kernel=(15, 15)) -> np.ndarray:\n    masked = img.copy()\n    mask_hide = (mask_vis == 0)\n    if mode == \"black\":\n        masked[mask_hide] = 0\n    elif mode == \"blur\":\n        blurred = cv2.GaussianBlur(img, blur_kernel, 0)\n        masked[mask_hide] = blurred[mask_hide]\n    elif mode == \"mix\":\n        alpha = 0.5\n        blurred = cv2.GaussianBlur(img, blur_kernel, 0)\n        masked = np.where(\n            mask_hide[:, :, None],\n            (alpha * blurred + (1 - alpha) * img).astype(np.uint8),\n            img,\n        )\n    return masked\n\nbase_augment = Compose([\n    OneOf([\n        Rotate(limit=15, border_mode=cv2.BORDER_REFLECT),\n        Affine(translate_percent={\"x\": (-0.1, 0.1), \"y\": (-0.1, 0.1)}, scale=(0.9, 1.1)),\n        Perspective(scale=(0.05, 0.15))\n    ], p=0.6),\n    OneOf([\n        RandomBrightnessContrast(brightness_limit=0.2, contrast_limit=0.2),\n        HueSaturationValue(hue_shift_limit=15, sat_shift_limit=20, val_shift_limit=15),\n    ], p=0.5),\n    OneOf([GaussianBlur(blur_limit=3), MotionBlur(blur_limit=5)], p=0.3),\n])\n\ndef resize_or_pad_to_target(img: np.ndarray, target_size: tuple[int, int]) -> np.ndarray:\n    \"\"\"Resize (or optionally pad/crop) to the target size.\"\"\"\n    h, w = img.shape[:2]\n    tw, th = target_size  # target width, height\n\n    # If already target size, nothing to do\n    if (w, h) == (tw, th):\n        return img\n\n    # Simple resize (may distort aspect ratio)\n    img_resized = cv2.resize(img, target_size, interpolation=cv2.INTER_LINEAR)\n    return img_resized\n\n    # If you prefer preserving aspect ratio with padding, you can implement that instead:\n    \"\"\"\n    scale = min(tw / w, th / h)\n    new_w = int(w * scale)\n    new_h = int(h * scale)\n    resized = cv2.resize(img, (new_w, new_h), interpolation=cv2.INTER_LINEAR)\n    # pad to target\n    pad_w = tw - new_w\n    pad_h = th - new_h\n    top = pad_h // 2\n    bottom = pad_h - top\n    left = pad_w // 2\n    right = pad_w - left\n    padded = cv2.copyMakeBorder(\n        resized, top, bottom, left, right,\n        borderType=cv2.BORDER_CONSTANT, value=[0, 0, 0]\n    )\n    return padded\n    \"\"\"\n\ndef augment_image_with_occlusion(img: np.ndarray) -> np.ndarray:\n    # First, optionally resize input to target size to standardize pipeline\n    # img = resize_or_pad_to_target(img, TARGET_SIZE)\n\n    n_trans = random.randint(MIN_TRANSFORMS, MAX_TRANSFORMS)\n    aug = img.copy()\n    for _ in range(n_trans):\n        aug = base_augment(image=aug)[\"image\"]\n    if random.random() < MASK_PROB:\n        label_map = get_label_map(aug)\n        face_mask = make_face_mask(label_map)\n        option = random.choice(MASK_OPTIONS)\n        mask_vis = get_region_mask_by_option(label_map, face_mask, option)\n        mode = random.choice(MASK_MODES)\n        aug = apply_mask_to_image(aug, mask_vis, mode=mode)\n\n    # Final resizing to ensure exactly target size\n    # aug = resize_or_pad_to_target(aug, TARGET_SIZE)\n    return aug\n\ndef augment_row(row,\n                img_dir=IMG_DIR,\n                aug_dir=AUG_DIR,\n                augment_rate=AUGMENT_RATE,\n                min_variants=MIN_VARIANTS,\n                max_variants=MAX_VARIANTS):\n    augmented_rows = []\n    augmented_rows.append({\n        \"img1\": row[\"img1\"],\n        \"img2\": row[\"img2\"],\n        \"label\": row[\"label\"]\n    })\n\n    if random.random() < augment_rate:\n        n_var = random.randint(min_variants, max_variants)\n        for v in range(n_var):\n            do1 = (random.random() < 0.5)\n            do2 = (random.random() < 0.5)\n            if not (do1 or do2):\n                do1 = True\n\n            new1 = row[\"img1\"]\n            new2 = row[\"img2\"]\n\n            if do1:\n                path1 = os.path.join(img_dir, row[\"img1\"])\n                img1 = cv2.imread(path1)\n                if img1 is not None:\n                    aug1 = augment_image_with_occlusion(img1)\n                    base, ext = os.path.splitext(row[\"img1\"])\n                    new_fn1 = f\"{base}_aug{v}{ext}\"\n                    os.makedirs(aug_dir, exist_ok=True)\n                    cv2.imwrite(os.path.join(aug_dir, new_fn1), aug1)\n                    new1 = new_fn1\n\n            if do2:\n                path2 = os.path.join(img_dir, row[\"img2\"])\n                img2 = cv2.imread(path2)\n                if img2 is not None:\n                    aug2 = augment_image_with_occlusion(img2)\n                    base, ext = os.path.splitext(row[\"img2\"])\n                    new_fn2 = f\"{base}_aug{v}{ext}\"\n                    os.makedirs(aug_dir, exist_ok=True)\n                    cv2.imwrite(os.path.join(aug_dir, new_fn2), aug2)\n                    new2 = new_fn2\n\n            augmented_rows.append({\n                \"img1\": new1,\n                \"img2\": new2,\n                \"label\": row[\"label\"]\n            })\n\n    return augmented_rows","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-16T22:30:36.966528Z","iopub.execute_input":"2025-10-16T22:30:36.966756Z","iopub.status.idle":"2025-10-16T22:30:56.058738Z","shell.execute_reply.started":"2025-10-16T22:30:36.966740Z","shell.execute_reply":"2025-10-16T22:30:56.058128Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def get_id_from_filename(fn):\n    # Ex : \"001_1.jpg\" ou \"001_2.png\" -> renvoie \"001\"\n    return os.path.splitext(fn)[0].rsplit(\"_\", 1)[0]\n\n# Lister tous les fichiers dans le dossier d’entraînement\ntrain_files = sorted([f for f in os.listdir(TRAIN_IMG_DIR) if os.path.isfile(os.path.join(TRAIN_IMG_DIR, f))])\n\n# Regrouper par ID\nids = {}\nfor fn in train_files:\n    id_ = get_id_from_filename(fn)\n    ids.setdefault(id_, []).append(fn)\n\n# Garde seulement les IDs avec exactement ou au moins 2 fichiers\nids = {k: v for k, v in ids.items() if len(v) >= 2}\nprint(\"Nombre d’IDs valides :\", len(ids))\n\n# Créer la liste des paires (img1, img2), sans label pour l’instant\npairs = []\nfor k, imgs in sorted(ids.items()):\n    imgs = sorted(imgs)\n    # On suppose que pour chaque ID il y a exactement 2 images (ex : \"001_1\", \"001_2\")\n    pairs.append((imgs[0], imgs[1]))\n\n# Maintenant appliquer le schéma : 200 premières paires → positives, 200 suivantes → négatives, etc.\npairs_labeled = []\ncycle = 400\nhalf = 200\n\nfor idx, (fn1, fn2) in enumerate(pairs):\n    mod = idx % cycle\n    if mod < half:\n        label = 1\n    else:\n        label = 0\n    pairs_labeled.append((fn1, fn2, label))\n\npairs_df = pd.DataFrame(pairs_labeled, columns=[\"img1\", \"img2\", \"label\"])\nprint(\"Total paires avec labels:\", len(pairs_df))\nprint(pairs_df[\"label\"].value_counts())\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-16T22:30:56.060700Z","iopub.execute_input":"2025-10-16T22:30:56.061146Z","iopub.status.idle":"2025-10-16T22:31:00.075099Z","shell.execute_reply.started":"2025-10-16T22:30:56.061127Z","shell.execute_reply":"2025-10-16T22:31:00.074309Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# train_df est ton DataFrame avec colonnes \"img1\", \"img2\", \"label\"\n\nfrom sklearn.model_selection import train_test_split\ntrain_df, val_df = train_test_split(pairs_df, test_size=0.15, stratify=pairs_df[\"label\"], random_state=42)\n\nlist_of_dfs = []\nfor _, row in train_df.iterrows():\n    list_of_dfs.append(pd.DataFrame(augment_row(row)))\n\ntrain_df = pd.concat(list_of_dfs, ignore_index=True)\ntrain_df = train_df.reset_index(drop=True)\n\n\n\n\n# 4. Split train / val etc.\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-16T22:31:00.075913Z","iopub.execute_input":"2025-10-16T22:31:00.076210Z","iopub.status.idle":"2025-10-16T22:33:20.020145Z","shell.execute_reply.started":"2025-10-16T22:31:00.076185Z","shell.execute_reply":"2025-10-16T22:33:20.019526Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"train_df","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-16T22:33:20.020869Z","iopub.execute_input":"2025-10-16T22:33:20.021079Z","iopub.status.idle":"2025-10-16T22:33:20.038833Z","shell.execute_reply.started":"2025-10-16T22:33:20.021062Z","shell.execute_reply":"2025-10-16T22:33:20.038106Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"train_df[\"label\"].value_counts()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-16T22:33:20.039718Z","iopub.execute_input":"2025-10-16T22:33:20.039966Z","iopub.status.idle":"2025-10-16T22:33:20.050194Z","shell.execute_reply.started":"2025-10-16T22:33:20.039947Z","shell.execute_reply":"2025-10-16T22:33:20.049540Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import matplotlib.pyplot as plt\nfrom PIL import Image\n\nimport os\nimport matplotlib.pyplot as plt\nfrom PIL import Image\n\ndef show_pairs_from_df(df_pairs, img_dir, aug_dir=None, n_pairs=10):\n    \"\"\"\n    Affiche les n_pairs premières entrées de df_pairs,\n    chaque paire (img1, img2) côte à côte.\n    Si une image n’est pas trouvée dans img_dir et qu’aug_dir est donné,\n    essaie de la charger depuis aug_dir.\n    \"\"\"\n    n = min(n_pairs, len(df_pairs))\n    fig, axes = plt.subplots(n, 2, figsize=(6, n * 3))\n    for i in range(n):\n        row = df_pairs.iloc[i]\n        fn1 = row[\"img1\"]\n        fn2 = row[\"img2\"]\n        lbl = row[\"label\"]\n\n        # fonction utilitaire pour charger image depuis img_dir ou aug_dir\n        def load_img(fn):\n            path1 = os.path.join(img_dir, fn)\n            if os.path.exists(path1):\n                return Image.open(path1).convert(\"RGB\")\n            if aug_dir is not None:\n                path2 = os.path.join(aug_dir, fn)\n                if os.path.exists(path2):\n                    return Image.open(path2).convert(\"RGB\")\n            # Si pas trouvé, lever erreur ou retourner une image vide\n            raise FileNotFoundError(f\"Image '{fn}' non trouvée ni dans '{img_dir}' ni dans '{aug_dir}'\")\n\n        img1 = load_img(fn1)\n        img2 = load_img(fn2)\n\n        ax1 = axes[i, 0]\n        ax2 = axes[i, 1]\n        ax1.imshow(img1)\n        ax1.axis(\"off\")\n        ax1.set_title(fn1)\n        ax2.imshow(img2)\n        ax2.axis(\"off\")\n        ax2.set_title(f\"{fn2}\\nLabel = {lbl}\")\n\n    plt.tight_layout()\n    plt.show()\n\n\n# Exemple d’utilisation avec train_df :\nshow_pairs_from_df(train_df, TRAIN_IMG_DIR, aug_dir=\"/kaggle/working/aug_images\", n_pairs=20)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-16T22:33:20.050899Z","iopub.execute_input":"2025-10-16T22:33:20.051138Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"MODEL_NAME = \"google/vit-base-patch16-224-in21k\"\nfeature_extractor = AutoFeatureExtractor.from_pretrained(MODEL_NAME)\nvit = AutoModel.from_pretrained(MODEL_NAME).to(device)\n\n# On peut geler le modèle ou certaines couches\nvit.eval()\nfor p in vit.parameters():\n    p.requires_grad = False\n\nprint(\"Loaded ViT backbone:\", MODEL_NAME)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class FacePairDataset(Dataset):\n    def __init__(self, df, img_dir, feature_extractor, transform=None):\n        self.df = df.reset_index(drop=True)\n        self.img_dir = img_dir\n        self.feature_extractor = feature_extractor\n        self.transform = transform\n\n    def __len__(self):\n        return len(self.df)\n\n    def load_image(self, fn):\n        path1 = os.path.join(self.img_dir, fn)\n        if os.path.exists(path1):\n            img = Image.open(path1).convert(\"RGB\")\n            return img\n        else :\n            path2 = os.path.join(\"/kaggle/working/aug_images\", fn)\n            img = Image.open(path2).convert(\"RGB\")\n            return img\n     \n\n    def __getitem__(self, idx):\n        row = self.df.iloc[idx]\n        img1 = self.load_image(row[\"img1\"])\n        img2 = self.load_image(row[\"img2\"])\n        if self.transform:\n            img1 = self.transform(img1)\n            img2 = self.transform(img2)\n        # feature_extractor expects a PIL image (or list), returns dict with 'pixel_values'\n        enc1 = self.feature_extractor(images=img1, return_tensors=\"pt\")\n        enc2 = self.feature_extractor(images=img2, return_tensors=\"pt\")\n        pv1 = enc1[\"pixel_values\"].squeeze(0)\n        pv2 = enc2[\"pixel_values\"].squeeze(0)\n        label = torch.tensor(row[\"label\"], dtype=torch.float)\n        return {\"pv1\": pv1, \"pv2\": pv2, \"label\": label}\n\n\nsize = (224, 224)\ntransform = T.Compose([\n    T.Resize(224),\n    T.RandomHorizontalFlip(p=0.5),\n    # éventuellement ColorJitter, etc.\n])\n\ntrain_dataset = FacePairDataset(train_df, TRAIN_IMG_DIR, feature_extractor, transform=transform)\nval_dataset   = FacePairDataset(val_df, TRAIN_IMG_DIR, feature_extractor, transform=T.Resize(size))\n\ntrain_loader = DataLoader(train_dataset, batch_size=300, shuffle=True, pin_memory=True)\nval_loader   = DataLoader(val_dataset, batch_size=64, shuffle=False,  pin_memory=True)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"@torch.no_grad()\ndef embed_from_pv(pv):\n    out = vit(pixel_values=pv.to(device))\n    # out.last_hidden_state shape (B, seq_len, hidden_size)\n    # Le token [CLS] est à l'index 0\n    cls = out.last_hidden_state[:, 0, :]\n    return cls.cpu()\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"with torch.no_grad():\n    dummy = torch.zeros((1, ) + train_dataset[0][\"pv1\"].shape).to(device)\n    emb_dim = embed_from_pv(dummy).shape[-1]\nprint(\"Embedding dimension:\", emb_dim)\n\nclass ComparisonHead(nn.Module):\n    def __init__(self, emb_dim, hidden_dim=256):\n        super().__init__()\n        self.net = nn.Sequential(\n            nn.Linear(emb_dim * 2, hidden_dim),\n            nn.ReLU(),\n            nn.BatchNorm1d(hidden_dim),\n            nn.Dropout(0.2),\n            nn.Linear(hidden_dim, 1)\n            \n        )\n    def forward(self, e1, e2):\n        x = torch.cat([e1, e2], dim=1)\n        return self.net(x).squeeze(1)\n\nhead = ComparisonHead(emb_dim).to(device)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-17T02:28:23.932857Z","iopub.execute_input":"2025-10-17T02:28:23.933494Z","iopub.status.idle":"2025-10-17T02:28:23.988605Z","shell.execute_reply.started":"2025-10-17T02:28:23.933459Z","shell.execute_reply":"2025-10-17T02:28:23.987778Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"criterion = nn.BCEWithLogitsLoss()\noptimizer = torch.optim.AdamW(head.parameters(), lr=1e-4, weight_decay=1e-4)\nscheduler = torch.optim.lr_scheduler.StepLR(optimizer, step_size=3, gamma=0.5)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-17T00:25:09.268137Z","iopub.execute_input":"2025-10-17T00:25:09.268628Z","iopub.status.idle":"2025-10-17T00:25:09.273046Z","shell.execute_reply.started":"2025-10-17T00:25:09.268606Z","shell.execute_reply":"2025-10-17T00:25:09.272360Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"best_auc = 0.0\nbest_path = \"best_head_vit.pt\"\n\nfor epoch in range(10):\n    head.train()\n    total_loss = 0.0\n    for batch in tqdm(train_loader, desc=f\"Train epoch {epoch+1}\"):\n        pv1 = batch[\"pv1\"]\n        pv2 = batch[\"pv2\"]\n        lbl = batch[\"label\"].to(device)\n        e1 = embed_from_pv(pv1)\n        e2 = embed_from_pv(pv2)\n        optimizer.zero_grad()\n        logits = head(e1.to(device), e2.to(device))\n\n        loss = criterion(logits, lbl)\n        loss.backward()\n        optimizer.step()\n        total_loss += loss.item() * e1.size(0)\n    scheduler.step()\n    avg_loss = total_loss / len(train_loader.dataset)\n    print(\"Train loss:\", avg_loss)\n\n    # validation\n    head.eval()\n    preds, trues = [], []\n    with torch.no_grad():\n        for batch in val_loader:\n            e1 = embed_from_pv(batch[\"pv1\"])\n            e2 = embed_from_pv(batch[\"pv2\"])\n            logits = head(e1.to(device), e2.to(device))\n            probs = torch.sigmoid(logits).cpu().numpy().tolist()\n            preds.extend(probs)\n            trues.extend(batch[\"label\"].numpy().tolist())\n    auc = roc_auc_score(trues, preds)\n    pred_bin = [1 if p >= 0.5 else 0 for p in preds]\n    acc = accuracy_score(trues, pred_bin)\n    print(f\"Val AUC: {auc:.4f}, Acc: {acc:.4f}\")\n    if auc > best_auc:\n        best_auc = auc\n        torch.save(head.state_dict(), best_path)\n        print(\"=> Best head saved\")\n\nprint(\"Best validation AUC:\", best_auc)\nhead.load_state_dict(torch.load(best_path, map_location=device))\nhead.eval()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-17T02:35:21.008330Z","iopub.execute_input":"2025-10-17T02:35:21.008904Z","iopub.status.idle":"2025-10-17T02:55:55.562675Z","shell.execute_reply.started":"2025-10-17T02:35:21.008878Z","shell.execute_reply":"2025-10-17T02:55:55.562021Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport torch\nfrom torch.utils.data import Dataset, DataLoader\nfrom PIL import Image\nfrom tqdm import tqdm\n\nfrom transformers import ViTImageProcessor, ViTModel\n\n# ========== Configurations ==========\n\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n\nHEAD_CHECKPOINT = \"/kaggle/working/best_head_vit.pt\"\nIMAGE_FOLDER = \"/kaggle/input/anip-reconnaissance-faciale-estimation-ages-ocr/dataset_tache_1/dataset_tache_1/test\"\nIMG_EXTS = {\".jpg\", \".jpeg\", \".png\", \".bmp\", \".tif\", \".tiff\"}\n\nBATCH_SIZE = 16\nNUM_WORKERS = 4\nPIN_MEMORY = True\n\n# ========== Modèles ==========\n\nprocessor = ViTImageProcessor.from_pretrained(\"google/vit-base-patch16-224-in21k\")\nvit_model = ViTModel.from_pretrained(\"google/vit-base-patch16-224-in21k\")\nvit_model.eval()\nvit_model.to(device)\n\n# Définis la tête; ici un exemple, remplace-la par ta vraie implémentation\n\n\n# ========== Dataset pour le chargement des images ==========\n\nclass ImageDataset(Dataset):\n    def __init__(self, image_paths, processor):\n        self.image_paths = image_paths\n        self.processor = processor\n\n    def __len__(self):\n        return len(self.image_paths)\n\n    def __getitem__(self, idx):\n        path = self.image_paths[idx]\n        img = Image.open(path).convert(\"RGB\")\n        inputs = self.processor(images=img, return_tensors=\"pt\")\n        # inputs sont des tensors de forme (1, …), on retire la dimension batch\n        inputs = {k: v.squeeze(0) for k, v in inputs.items()}\n        return path, inputs\n\ndef collect_image_paths(folder):\n    paths = []\n    for root, dirs, files in os.walk(folder):\n        for fname in files:\n            _, ext = os.path.splitext(fname.lower())\n            if ext in IMG_EXTS:\n                paths.append(os.path.join(root, fname))\n    return paths\n\n# ========== Extraction des embeddings en lot ==========\n\ndef extract_all_embeddings(image_paths):\n    dataset = ImageDataset(image_paths, processor)\n    loader = DataLoader(dataset,\n                        batch_size=BATCH_SIZE,\n                        shuffle=False,\n                        num_workers=NUM_WORKERS,\n                        pin_memory=PIN_MEMORY,\n                        collate_fn=lambda batch: batch)\n    emb_cache = {}\n    with torch.no_grad():\n        for batch in tqdm(loader, desc=\"Extract embeddings\"):\n            paths = [item[0] for item in batch]\n            # on doit regrouper les inputs pour les passer au modèle\n            batched = {}\n            for k in batch[0][1].keys():\n                batched[k] = torch.stack([item[1][k] for item in batch], dim=0).to(device)\n            outputs = vit_model(**batched)\n            cls_emb = outputs.last_hidden_state[:, 0, :]  # token [CLS]\n            for i, p in enumerate(paths):\n                emb_cache[p] = cls_emb[i].detach().cpu()\n    return emb_cache\n\n# ========== Comparaison via la tête ==========\n\ndef find_best_matches_from_embeddings(emb_cache):\n    image_paths = list(emb_cache.keys())\n    n = len(image_paths)\n    emb_list = [emb_cache[p].to(device) for p in image_paths]\n    emb_tensor = torch.stack(emb_list, dim=0)  # shape (n, dim)\n\n    best_matches = {}\n    with torch.no_grad():\n        for i in tqdm(range(n), desc=\"Finding best match pairs\"):\n            e1 = emb_tensor[i : i + 1, :]  # (1, dim)\n            e1_expand = e1.repeat(n, 1)\n            e2_all = emb_tensor\n            logits = head(e1_expand, e2_all) \n            torch.sigmoid(logits).cpu().numpy().tolist()\n            # print(\"logits.shape:\", logits.shape)# (n, 1)\n            probs = torch.sigmoid(logits)  # (n,)\n            # print(probs)\n            probs[i] = -1.0  # éviter le self\n\n            max_idx = torch.argmax(probs).item()\n            best_prob = probs[max_idx].item()\n            best_path = image_paths[max_idx]\n            best_matches[image_paths[i]] = (best_path, best_prob)\n    return best_matches\n\n# ========== Script principal ==========\ndim_embedding = vit_model.config.hidden_size\nhead.load_state_dict(torch.load(HEAD_CHECKPOINT, map_location=device))\nhead.eval()\n\nimage_paths = collect_image_paths(IMAGE_FOLDER)\nif len(image_paths) < 2:\n    print(\"Pas assez d’images pour comparer.\")\n\n# 1. extraire tous les embeddings\nemb_cache = extract_all_embeddings(image_paths)\n\n# 2. trouver les meilleures correspondances\nbest_matches = find_best_matches_from_embeddings(emb_cache)\n\n\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-17T02:25:30.963906Z","iopub.execute_input":"2025-10-17T02:25:30.964564Z","iopub.status.idle":"2025-10-17T02:25:56.246818Z","shell.execute_reply.started":"2025-10-17T02:25:30.964535Z","shell.execute_reply":"2025-10-17T02:25:56.245985Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\n\n# best_matches est : dict image → (best_match, prob)\n# Création d’un DataFrame\ndata = []\nfor img, (match, prob) in best_matches.items():\n    data.append({\n        \"image\": img,\n        \"best_match\": match,\n        \"probability\": prob\n    })\n\ndf = pd.DataFrame(data)\n# Optionnel : trier par probabilité décroissante\n\nprint(df.head())  # affiche les 5 premières lignes\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-17T02:26:23.115325Z","iopub.execute_input":"2025-10-17T02:26:23.115598Z","iopub.status.idle":"2025-10-17T02:26:23.125920Z","shell.execute_reply.started":"2025-10-17T02:26:23.115578Z","shell.execute_reply":"2025-10-17T02:26:23.125078Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import matplotlib.pyplot as plt\nfrom PIL import Image\n\ndef show_pair_from_df(row):\n    img_path = row[\"image\"]\n    match_path = row[\"best_match\"]\n    prob = row[\"probability\"]\n\n    img1 = Image.open(img_path).convert(\"RGB\")\n    img2 = Image.open(match_path).convert(\"RGB\")\n\n    fig, axes = plt.subplots(1, 2, figsize=(8, 4))\n    axes[0].imshow(img1)\n    axes[0].set_title(\"Image\")\n    axes[0].axis(\"off\")\n\n    axes[1].imshow(img2)\n    axes[1].set_title(f\"Match (prob = {prob:.3f})\")\n    axes[1].axis(\"off\")\n\n    plt.tight_layout()\n    plt.show()\n\n# Exemple : afficher les top 5 paires les plus probables\nfor idx, row in df.head(5).iterrows():\n    show_pair_from_df(row)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-17T02:26:36.964179Z","iopub.execute_input":"2025-10-17T02:26:36.964837Z","iopub.status.idle":"2025-10-17T02:26:38.331281Z","shell.execute_reply.started":"2025-10-17T02:26:36.964808Z","shell.execute_reply":"2025-10-17T02:26:38.330095Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from numpy.linalg import norm\ndef cos_sim(a, b):\n    return float(np.dot(a, b) / (norm(a)*norm(b) + 1e-8))\n\ndef head_score(a, b):\n    ta = torch.tensor(a, dtype=torch.float32).unsqueeze(0).to(device)\n    tb = torch.tensor(b, dtype=torch.float32).unsqueeze(0).to(device)\n    with torch.no_grad():\n        logit = head(ta, tb).item()\n    return 1 / (1 + np.exp(-logit))\n\nmatches = []\nfor tfn, temb in test_emb.items():\n    sims = [(fn, cos_sim(temb, train_emb[fn])) for fn in train_emb.keys()]\n    best_fn, best_sim = max(sims, key=lambda x: x[1])\n    prob = head_score(temb, train_emb[best_fn])\n    matches.append({\n        \"test_image\": tfn,\n        \"train_image\": best_fn,\n        \"train_id\": get_id_from_filename(best_fn),\n        \"cosine\": best_sim,\n        \"head_prob\": prob\n    })\ndf_match = pd.DataFrame(matches)\ndf_match.to_csv(\"submission_vit_test2train.csv\", index=False)\nprint(\"Submission file generated: submission_vit_test2train.csv\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}