{"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"codemirror_mode":{"name":"ipython","version":3},"file_extension":".py","mimetype":"text/x-python","name":"python","nbconvert_exporter":"python","pygments_lexer":"ipython3","version":"3.12.12"},"kaggle":{"accelerator":"nvidiaTeslaT4","dataSources":[{"sourceId":129750,"databundleVersionId":15563344,"sourceType":"competition"}],"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# RF-DETR (robust) — sequential train → infer → combine → submit\n\nThis notebook is reorganized to keep **all imports + utilities at the start**, then run **Face** end-to-end, then **Shemagh**, then combine + tune + submit + visualize.","metadata":{}},{"cell_type":"code","source":"!pip -q install -U rfdetr supervision pycocotools albumentations opencv-python","metadata":{"execution":{"iopub.status.busy":"2026-02-10T01:32:41.567723Z","iopub.execute_input":"2026-02-10T01:32:41.567967Z","iopub.status.idle":"2026-02-10T01:32:57.083660Z","shell.execute_reply.started":"2026-02-10T01:32:41.567927Z","shell.execute_reply":"2026-02-10T01:32:57.082997Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Phase 1 — Imports & global config","metadata":{}},{"cell_type":"code","source":"# =========================\n# Imports (keep them here; avoid scattered imports later)\n# =========================\nimport os, json, math, random, shutil, inspect, argparse\nfrom pathlib import Path\nfrom dataclasses import dataclass\nfrom typing import Optional, Dict, Any, List, Tuple\nimport numpy as np\nimport pandas as pd\nfrom sklearn.model_selection import train_test_split\nfrom PIL import Image\nimport matplotlib.pyplot as plt\nimport matplotlib.patches as patches\nfrom matplotlib.patches import Rectangle\nimport torch\nfrom torch.serialization import safe_globals\nfrom IPython.display import display\nimport cv2\nimport albumentations as A\nfrom rfdetr import RFDETRSmall, RFDETRMedium, RFDETRLarge, RFDETRBase\nimport os, json, shutil, random, math, warnings, inspect\nfrom pathlib import Path\n\nimport numpy as np\nimport pandas as pd\n\nfrom PIL import Image\n\nwarnings.filterwarnings(\"ignore\")\n\n# --------- Repro ---------\nSEED = 42\n\ndef set_seed(seed: int = 42):\n    os.environ[\"PYTHONHASHSEED\"] = str(seed)\n    random.seed(seed)\n    np.random.seed(seed)\n    try:\n        import torch\n        torch.manual_seed(seed)\n        if torch.cuda.is_available():\n            torch.cuda.manual_seed_all(seed)\n        torch.backends.cudnn.deterministic = True\n        torch.backends.cudnn.benchmark = False\n    except Exception:\n        pass\n\nset_seed(SEED)\n\n# --------- Paths (Kaggle) ---------\nROOT = Path(\"/kaggle/input/dal-shemagh-detection-challenge\")\nIMAGES_DIR = ROOT / \"images\" / \"train\"\nLABELS_DIR = ROOT / \"labels\" / \"train\"\nCSV_PATH   = ROOT / \"train_labels.csv\"\n\nTEST_IMAGES_DIR = ROOT / \"images\" / \"test\"\n\n# Outputs\nYOLO_SPLIT_DIR   = Path(\"yolo_split_80_20\")\nCOCO_DATASET_DIR = Path(\"rfdetr_dataset\")              # base COCO (train/valid/test)\nAUG_DATASET_DIR  = Path(\"/kaggle/working/rfdetr_dataset_aug\")  # augmented copy (train augmented only)\n\n# Two single-class datasets (share the same images via symlinks when possible)\nFACE_DATASET_DIR = Path(\"/kaggle/working/rfdetr_dataset_face\")\nSHEM_DATASET_DIR = Path(\"/kaggle/working/rfdetr_dataset_shemagh\")\n\n# Model outputs\nOUT_FACE = Path(\"rfdetr_output_face\")\nOUT_SHEM = Path(\"rfdetr_output_shemagh\")\n\n# --------- Class conventions ---------\n# Competition YOLO label ids:\nFACE_YOLO = 0\nSHEM_YOLO = 1\nNAMES = [\"face\", \"shemagh\"]\n\n# COCO category ids we write for the multi-class COCO dataset:\nYOLO_TO_COCO = {0: 1, 1: 2}\nCOCO_TO_YOLO = {1: 0, 2: 1}\n\n# If you visually confirm the dataset has swapped labels, set this to True.\nSWAP_RAW_YOLO_LABELS = False\n\ndef remap_yolo_cls(cls: int) -> int:\n    cls = int(cls)\n    return (1 - cls) if (SWAP_RAW_YOLO_LABELS and cls in (0, 1)) else cls\n\n# --------- Augmentation knobs ---------\nAUG_RATIO   = 2.0   # e.g. 2.0 => +2x extra augmented images\nMIXUP_RATIO = 0.05   # keep small; strong augmentation\nUSE_GEOM    = True\nGEOM_PROB   = 0.2   # fraction of augmented samples using geometry ops\n\n# --------- Training knobs ---------\n# Tip: RF-DETR prefers resolution divisible by 56. Use 448/560/672 ...\nMODEL_RESOLUTION = 896  # safe default; try 448/560 if VRAM allows\nEPOCHS = 10\nBATCH_SIZE = 2\nGRAD_ACCUM = 32\nLR = 2e-4\n\n# --------- Pretrained initialization ---------\n# Used to initialize both single-class models (FACE + SHEMAGH)\nINIT_PRETRAIN_WEIGHTS = Path(\"rf-detr-large-2026.pth\")\nif not INIT_PRETRAIN_WEIGHTS.exists():\n    # Try to locate it under /kaggle/input if you attached it as a Kaggle dataset\n    _cands = sorted(Path(\"/kaggle/input\").glob(\"**/rf-detr-large-2026.pth\"))\n    INIT_PRETRAIN_WEIGHTS = _cands[0] if _cands else None\n\n# --------- Inference (TTA) ---------\nUSE_TTA = True  # set False to disable test-time augmentation\n\n\n# --------- Inference knobs ---------\nCONF_FACE = 0.10\nCONF_SHEM = 0.10\nNMS_IOU   = 0.65\n\n# Containment tuning grid\nCONTAIN_GRID = np.linspace(0.0, 1.0, 101)\n\ndef _ensure_clean_dir(p: Path):\n    if p.exists():\n        shutil.rmtree(p)\n    p.mkdir(parents=True, exist_ok=True)\n\ndef _safe_symlink(src: Path, dst: Path, prefer_symlink: bool = True):\n    '''\n    Creates a robust link from dst -> src.\n\n    Why this exists:\n    - If src was passed as a *relative* path, the symlink becomes relative to dst's folder,\n      which easily produces BROKEN symlinks.\n    - If a broken symlink already exists, Path.exists() returns False but the dir entry still exists.\n\n    Strategy:\n    1) Resolve src to an absolute path.\n    2) If dst is an existing *broken* symlink, remove it.\n    3) Try to symlink to absolute target; if that fails, fall back to copying.\n    '''\n    src = Path(src)\n    dst = Path(dst)\n    dst.parent.mkdir(parents=True, exist_ok=True)\n\n    if dst.exists():\n        return\n\n    if dst.is_symlink() and (not dst.exists()):\n        try:\n            dst.unlink()\n        except Exception:\n            pass\n\n    src_abs = src.resolve()\n\n    if prefer_symlink:\n        try:\n            os.symlink(str(src_abs), str(dst))\n            return\n        except Exception:\n            pass\n\n    try:\n        shutil.copy2(str(src_abs), str(dst))\n    except Exception:\n        if src_abs.is_dir():\n            shutil.copytree(src_abs, dst, dirs_exist_ok=True)\n        else:\n            shutil.copy2(src_abs, dst)\n\ndef _list_images(d: Path):\n    exts = {\".jpg\", \".jpeg\", \".png\", \".webp\"}\n    return sorted([p for p in d.iterdir() if p.suffix.lower() in exts])\n\nprint(\"Config OK.\")\n\ndef _safe_copy(src: Path, dst: Path):\n    src = Path(src).resolve()\n    dst = Path(dst)\n    dst.parent.mkdir(parents=True, exist_ok=True)\n    if dst.exists():\n        return\n    shutil.copy2(str(src), str(dst))\n\ndef _assert_coco_split_has_files(split_dir: Path):\n    '''\n    Asserts that every COCO images[i].file_name exists under split_dir.\n    Useful to catch issues before RF-DETR starts training.\n    '''\n    split_dir = Path(split_dir)\n    ann_path = split_dir / \"_annotations.coco.json\"\n    coco = json.loads(ann_path.read_text())\n    missing = []\n    for im in coco.get(\"images\", []):\n        fp = split_dir / im[\"file_name\"]\n        if not fp.exists():\n            missing.append(im[\"file_name\"])\n    if missing:\n        raise FileNotFoundError(f\"COCO split has {len(missing)} missing files (e.g. {missing[:5]}) at {split_dir}\")\n","metadata":{"execution":{"iopub.status.busy":"2026-02-10T01:32:57.085147Z","iopub.execute_input":"2026-02-10T01:32:57.085408Z","iopub.status.idle":"2026-02-10T01:33:34.162600Z","shell.execute_reply.started":"2026-02-10T01:32:57.085378Z","shell.execute_reply":"2026-02-10T01:33:34.161993Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Phase 2 — Dataset split + COCO conversion","metadata":{}},{"cell_type":"code","source":"# =========================\n# 1) Build stratified YOLO train/val split\n# =========================\n\ndef _count_bins(n: int) -> int:\n    if n == 0: return 0\n    if n == 1: return 1\n    if n == 2: return 2\n    return 3  # 3+\n\ndef _safe_read_label_counts(label_path: Path, face_id=0, shem_id=1):\n    n_face = 0\n    n_shem = 0\n    if label_path.exists():\n        txt = label_path.read_text(encoding=\"utf-8\").strip().splitlines()\n        for line in txt:\n            line = line.strip()\n            if not line:\n                continue\n            cls = remap_yolo_cls(int(float(line.split()[0])))\n            if cls == face_id:\n                n_face += 1\n            elif cls == shem_id:\n                n_shem += 1\n    return n_face, n_shem\n\ndef build_yolo_split(out_root: Path, test_size: float = 0.2, seed: int = 42):\n    if not CSV_PATH.exists():\n        raise FileNotFoundError(f\"Missing CSV: {CSV_PATH}. Are you running on Kaggle with the dataset attached?\")\n    if not IMAGES_DIR.exists() or not LABELS_DIR.exists():\n        raise FileNotFoundError(\"Train images/labels folders not found. Check ROOT path.\")\n\n    df = pd.read_csv(CSV_PATH)\n    if \"filename\" not in df.columns or \"right_place\" not in df.columns:\n        raise ValueError(\"train_labels.csv must contain columns: filename, right_place\")\n\n    df[\"filename\"] = df[\"filename\"].astype(str)\n    df[\"right_place\"] = df[\"right_place\"].astype(bool).astype(int)\n\n    n_face_list, n_shem_list = [], []\n    missing_images = []\n    for fn in df[\"filename\"].tolist():\n        img_path = IMAGES_DIR / fn\n        if not img_path.exists():\n            missing_images.append(fn)\n        stem = Path(fn).stem\n        lab_path = LABELS_DIR / f\"{stem}.txt\"\n        nf, ns = _safe_read_label_counts(lab_path, face_id=FACE_YOLO, shem_id=SHEM_YOLO)\n        n_face_list.append(nf)\n        n_shem_list.append(ns)\n\n    if missing_images:\n        raise FileNotFoundError(f\"Missing {len(missing_images)} train images. Example: {missing_images[:5]}\")\n    df[\"n_face\"] = n_face_list\n    df[\"n_shem\"] = n_shem_list\n\n    df[\"face_bin\"] = df[\"n_face\"].map(_count_bins)\n    df[\"shem_bin\"] = df[\"n_shem\"].map(_count_bins)\n\n    df[\"stratum_fine\"] = (\n        df[\"right_place\"].astype(str) + \"|\" +\n        df[\"face_bin\"].astype(str) + \"|\" +\n        df[\"shem_bin\"].astype(str)\n    )\n\n    vc = df[\"stratum_fine\"].value_counts()\n    rare = set(vc[vc < 2].index)\n\n    df[\"has_face\"] = (df[\"n_face\"] > 0).astype(int)\n    df[\"has_shem\"] = (df[\"n_shem\"] > 0).astype(int)\n    df[\"stratum_mid\"] = (\n        df[\"right_place\"].astype(str) + \"|\" +\n        df[\"has_face\"].astype(str) + \"|\" +\n        df[\"has_shem\"].astype(str)\n    )\n\n    df[\"stratum\"] = df[\"stratum_fine\"]\n    df.loc[df[\"stratum\"].isin(rare), \"stratum\"] = df.loc[df[\"stratum\"].isin(rare), \"stratum_mid\"]\n    vc2 = df[\"stratum\"].value_counts()\n    rare2 = set(vc2[vc2 < 2].index)\n    df.loc[df[\"stratum\"].isin(rare2), \"stratum\"] = df.loc[df[\"stratum\"].isin(rare2), \"right_place\"].astype(str)\n\n    train_df, val_df = train_test_split(\n        df, test_size=test_size, random_state=seed, stratify=df[\"stratum\"]\n    )\n\n    print(\"Train:\", len(train_df), \"Val:\", len(val_df))\n    print(\"right_place mean - Train:\", train_df[\"right_place\"].mean(), \"Val:\", val_df[\"right_place\"].mean())\n\n    _ensure_clean_dir(out_root)\n    (out_root / \"images/train\").mkdir(parents=True, exist_ok=True)\n    (out_root / \"images/val\").mkdir(parents=True, exist_ok=True)\n    (out_root / \"labels/train\").mkdir(parents=True, exist_ok=True)\n    (out_root / \"labels/val\").mkdir(parents=True, exist_ok=True)\n\n    def _copy_split(split_df, img_out, lab_out):\n        copied_img = copied_lab = missing_lab = 0\n        for fn in split_df[\"filename\"].tolist():\n            src_img = IMAGES_DIR / fn\n            dst_img = img_out / fn\n            dst_img.parent.mkdir(parents=True, exist_ok=True)\n            shutil.copy2(src_img, dst_img)\n            copied_img += 1\n\n            stem = Path(fn).stem\n            src_lab = LABELS_DIR / f\"{stem}.txt\"\n            dst_lab = lab_out / f\"{stem}.txt\"\n            if src_lab.exists():\n                # optionally remap class ids (swap) at copy-time\n                txt = src_lab.read_text(encoding=\"utf-8\").strip().splitlines()\n                out_lines = []\n                for line in txt:\n                    line = line.strip()\n                    if not line:\n                        continue\n                    parts = line.split()\n                    if len(parts) < 5:\n                        continue\n                    cls = remap_yolo_cls(int(float(parts[0])))\n                    out_lines.append(\" \".join([str(cls)] + parts[1:5]))\n                dst_lab.write_text(\"\\n\".join(out_lines) + (\"\\n\" if out_lines else \"\"), encoding=\"utf-8\")\n                copied_lab += 1\n            else:\n                dst_lab.write_text(\"\", encoding=\"utf-8\")\n                missing_lab += 1\n        return copied_img, copied_lab, missing_lab\n\n    tr_i, tr_l, tr_m = _copy_split(train_df, out_root/\"images/train\", out_root/\"labels/train\")\n    va_i, va_l, va_m = _copy_split(val_df,   out_root/\"images/val\",   out_root/\"labels/val\")\n\n    print(f\"Train copied images={tr_i}, labels={tr_l}, missing_labels_created_empty={tr_m}\")\n    print(f\"Val   copied images={va_i}, labels={va_l}, missing_labels_created_empty={va_m}\")\n\n    train_df[[\"filename\",\"right_place\"]].to_csv(out_root / \"train_labels_split.csv\", index=False)\n    val_df[[\"filename\",\"right_place\"]].to_csv(out_root / \"val_labels_split.csv\", index=False)\n\n    print(\"Wrote:\", out_root / \"train_labels_split.csv\")\n    print(\"Wrote:\", out_root / \"val_labels_split.csv\")\n\nbuild_yolo_split(YOLO_SPLIT_DIR, test_size=0.2, seed=SEED)","metadata":{"execution":{"iopub.status.busy":"2026-02-10T01:33:34.163482Z","iopub.execute_input":"2026-02-10T01:33:34.163792Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# =========================\n# (Optional) Quick sanity check: visualize labels on a few images\n# If labels look swapped, set SWAP_RAW_YOLO_LABELS=True in the config cell and re-run from Cell 1.\n# =========================\n\ndef yolo_to_xyxy(x, y, w, h, W, H):\n    xc, yc, bw, bh = x*W, y*H, w*W, h*H\n    x1 = xc - bw/2.0\n    y1 = yc - bh/2.0\n    x2 = xc + bw/2.0\n    y2 = yc + bh/2.0\n    return x1, y1, x2, y2\n\ndef plot_yolo_split_samples(split=\"val\", k=9):\n    labels_csv = YOLO_SPLIT_DIR / (\"train_labels_split.csv\" if split == \"train\" else \"val_labels_split.csv\")\n    df = pd.read_csv(labels_csv)\n    img_dir = YOLO_SPLIT_DIR / f\"images/{split}\"\n    lab_dir = YOLO_SPLIT_DIR / f\"labels/{split}\"\n\n    df = df.sample(min(k, len(df)), random_state=SEED).reset_index(drop=True)\n\n    cols = 3\n    rows = int(math.ceil(len(df) / cols))\n    fig, axes = plt.subplots(rows, cols, figsize=(16, 5*rows))\n    axes = np.array(axes).reshape(rows, cols)\n\n    color = {0: \"#00FF66\", 1: \"#FF3B30\"}\n\n    for i, row in df.iterrows():\n        ax = axes[i//cols, i%cols]\n        fn = row[\"filename\"]\n        img_path = img_dir / fn\n        lab_path = lab_dir / f\"{Path(fn).stem}.txt\"\n\n        im = Image.open(img_path).convert(\"RGB\")\n        W, H = im.size\n        ax.imshow(im)\n        ax.axis(\"off\")\n        ax.set_title(f\"{fn} | right_place={int(row['right_place'])}\")\n\n        if lab_path.exists():\n            for line in lab_path.read_text().strip().splitlines():\n                parts = line.split()\n                if len(parts) < 5:\n                    continue\n                cls = int(float(parts[0]))\n                xc, yc, w, h = map(float, parts[1:5])\n                x1,y1,x2,y2 = yolo_to_xyxy(xc,yc,w,h,W,H)\n                ax.add_patch(patches.Rectangle((x1,y1), x2-x1, y2-y1, fill=False, linewidth=3, edgecolor=color.get(cls, \"#FFD60A\")))\n                ax.text(x1, max(0,y1-3), NAMES[cls], color=\"white\", fontsize=10,\n                        bbox=dict(facecolor=color.get(cls, \"#FFD60A\"), alpha=0.65, edgecolor=\"none\", boxstyle=\"round,pad=0.2\"))\n\n    # hide unused\n    for j in range(len(df), rows*cols):\n        axes[j//cols, j%cols].axis(\"off\")\n    plt.tight_layout()\n    plt.show()\n\nplot_yolo_split_samples(split=\"val\", k=9)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# =========================\n# 2) Convert YOLO split -> COCO dataset for RF-DETR\n# =========================\ndef yolo_label_to_coco_bbox(line, W, H):\n    t = line.strip().split()\n    if len(t) < 5:\n        return None\n    cls = remap_yolo_cls(int(float(t[0])))\n    xc = float(t[1]) * W\n    yc = float(t[2]) * H\n    bw = float(t[3]) * W\n    bh = float(t[4]) * H\n    x = xc - bw/2.0\n    y = yc - bh/2.0\n    x = max(0.0, x); y = max(0.0, y)\n    bw = max(0.0, min(bw, W - x))\n    bh = max(0.0, min(bh, H - y))\n    return cls, [x, y, bw, bh]\n\ndef build_coco_for_split(split_name, images_dir, labels_dir, out_dir, has_labels=True):\n    out_dir.mkdir(parents=True, exist_ok=True)\n    ann_path = out_dir / \"_annotations.coco.json\"\n\n    images = []\n    annotations = []\n    ann_id = 1\n    img_id = 1\n\n    for img_path in _list_images(images_dir):\n        dst_img = out_dir / img_path.name\n        if not dst_img.exists():\n            shutil.copy2(img_path, dst_img)\n\n        with Image.open(img_path) as im:\n            W, H = im.size\n\n        images.append({\"id\": img_id, \"file_name\": img_path.name, \"width\": W, \"height\": H})\n\n        if has_labels:\n            lab_path = labels_dir / (img_path.stem + \".txt\")\n            if lab_path.exists():\n                for line in lab_path.read_text(encoding=\"utf-8\").splitlines():\n                    line = line.strip()\n                    if not line:\n                        continue\n                    res = yolo_label_to_coco_bbox(line, W, H)\n                    if res is None:\n                        continue\n                    cls, bbox = res\n                    cat_id = YOLO_TO_COCO.get(cls, int(cls) + 1)\n                    annotations.append({\n                        \"id\": ann_id,\n                        \"image_id\": img_id,\n                        \"category_id\": int(cat_id),\n                        \"bbox\": bbox,\n                        \"area\": float(bbox[2] * bbox[3]),\n                        \"iscrowd\": 0,\n                    })\n                    ann_id += 1\n\n        img_id += 1\n\n    categories = [{\"id\": 1, \"name\": \"face\", \"supercategory\": \"object\"},\n                  {\"id\": 2, \"name\": \"shemagh\", \"supercategory\": \"object\"}]\n\n    coco = {\n        \"info\": {\"description\": f\"RF-DETR dataset ({split_name})\"},\n        \"licenses\": [],\n        \"images\": images,\n        \"annotations\": annotations,\n        \"categories\": categories,\n    }\n\n    with open(ann_path, \"w\") as f:\n        json.dump(coco, f)\n\n    print(f\"[{split_name}] images={len(images)} annotations={len(annotations)} -> {ann_path}\")\n    return ann_path\n\n# rebuild\nif COCO_DATASET_DIR.exists():\n    shutil.rmtree(COCO_DATASET_DIR)\nCOCO_DATASET_DIR.mkdir(parents=True, exist_ok=True)\n\nbuild_coco_for_split(\"train\", YOLO_SPLIT_DIR/\"images/train\", YOLO_SPLIT_DIR/\"labels/train\", COCO_DATASET_DIR/\"train\", has_labels=True)\nbuild_coco_for_split(\"valid\", YOLO_SPLIT_DIR/\"images/val\",   YOLO_SPLIT_DIR/\"labels/val\",   COCO_DATASET_DIR/\"valid\", has_labels=True)\n\n# test split: no labels in Kaggle, but RF-DETR expects a COCO json. Create empty annotations.\nbuild_coco_for_split(\"test\", TEST_IMAGES_DIR, Path(\"__no_labels__\"), COCO_DATASET_DIR/\"test\", has_labels=False)\n\nprint(\"COCO_DATASET_DIR ready:\", COCO_DATASET_DIR.resolve())","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Ensure Albumentations is available even if you execute cells out of order.\ntry:\n    A  # noqa: F821\nexcept NameError:\n    import albumentations as A\n\n# =========================\n# 3) (Optional) Augment TRAIN split ONCE (shared by both models)\n# =========================\n\ndef _bgr_rotate(img, **kwargs):\n    if img is None:\n        return img\n    if img.ndim == 3 and img.shape[2] >= 3:\n        perms = [(0,1,2),(0,2,1),(1,0,2),(1,2,0),(2,0,1),(2,1,0)]\n        p = perms[np.random.randint(len(perms))]\n        out = img.copy()\n        out[:, :, :3] = img[:, :, p]\n        return out\n    return img\n\ndef _make_bbox_params():\n    kwargs = dict(format=\"coco\", label_fields=[\"category_ids\"], min_visibility=0.10)\n    try:\n        sig = inspect.signature(A.BboxParams)\n        if \"clip\" in sig.parameters:\n            kwargs[\"clip\"] = True\n        if \"filter_invalid_bboxes\" in sig.parameters:\n            kwargs[\"filter_invalid_bboxes\"] = True\n    except Exception:\n        pass\n    return A.BboxParams(**kwargs)\n\ndef _make_color_pipeline():\n    return A.Compose(\n        [\n            A.HueSaturationValue(p=0.70),\n            A.RandomBrightnessContrast(p=0.70),\n            A.RandomGamma(p=0.45),\n            A.GaussNoise(p=0.40),\n            A.Lambda(image=_bgr_rotate, p=0.25),\n            A.CoarseDropout(max_holes=6, max_height=32, max_width=32, p=0.20),\n        ],\n        bbox_params=_make_bbox_params(),\n    )\n\ndef _make_geom_pipeline():\n    return A.Compose(\n        [\n            A.HueSaturationValue(p=0.70),\n            A.RandomBrightnessContrast(p=0.70),\n            A.RandomGamma(p=0.45),\n            A.GaussNoise(p=0.40),\n            A.Lambda(image=_bgr_rotate, p=0.25),\n            A.CoarseDropout(max_holes=6, max_height=32, max_width=32, p=0.20),\n\n            A.HorizontalFlip(p=0.50),\n            A.ShiftScaleRotate(\n                shift_limit=0.05,\n                scale_limit=0.15,\n                rotate_limit=12,\n                border_mode=cv2.BORDER_CONSTANT,\n                value=0,\n                p=0.35,\n            ),\n        ],\n        bbox_params=_make_bbox_params(),\n    )\n\nAUG_COLOR = _make_color_pipeline()\nAUG_GEOM  = _make_geom_pipeline()\n\ndef _clip_coco_bbox(b, W, H):\n    x, y, w, h = [float(v) for v in b]\n    if W <= 1 or H <= 1:\n        return None\n    x = max(0.0, min(x, W - 1.0))\n    y = max(0.0, min(y, H - 1.0))\n    w = max(0.0, min(w, W - x))\n    h = max(0.0, min(h, H - y))\n    if w < 1.0 or h < 1.0:\n        return None\n    return [x, y, w, h]\n\ndef _apply_aug(img_path, coco_bboxes, category_ids, out_path, pipeline):\n    img = cv2.imread(str(img_path))\n    if img is None:\n        return None, None\n\n    H, W = img.shape[:2]\n    b2, c2 = [], []\n    for b, cid in zip(coco_bboxes, category_ids):\n        bb = _clip_coco_bbox(b, W, H)\n        if bb is not None:\n            b2.append(bb)\n            c2.append(int(cid))\n\n    try:\n        t = pipeline(image=img, bboxes=b2, category_ids=c2)\n    except Exception:\n        t = AUG_COLOR(image=img, bboxes=b2, category_ids=c2)\n\n    img2 = t[\"image\"]\n    bboxes2 = t.get(\"bboxes\", [])\n    cids2   = t.get(\"category_ids\", [])\n\n    H2, W2 = img2.shape[:2]\n    b3, c3 = [], []\n    for b, cid in zip(bboxes2, cids2):\n        bb = _clip_coco_bbox(b, W2, H2)\n        if bb is not None:\n            b3.append(bb)\n            c3.append(int(cid))\n\n    out_path.parent.mkdir(parents=True, exist_ok=True)\n    cv2.imwrite(str(out_path), img2)\n    return b3, c3\n\ndef _mixup(imgA, bA, cA, imgB, bB, cB, alpha=0.4):\n    lam = np.random.beta(alpha, alpha)\n    mixed = (lam * imgA.astype(np.float32) + (1 - lam) * imgB.astype(np.float32)).clip(0, 255).astype(np.uint8)\n    H, W = mixed.shape[:2]\n    b_all, c_all = [], []\n    for b, cid in list(zip(bA, cA)) + list(zip(bB, cB)):\n        bb = _clip_coco_bbox(b, W, H)\n        if bb is not None:\n            b_all.append(bb)\n            c_all.append(int(cid))\n    return mixed, b_all, c_all\n\ndef augment_train_coco(src_dataset_dir: Path, dst_dataset_dir: Path, aug_ratio=0.0, mixup_ratio=0.0):\n    set_seed(SEED)\n\n    src_dataset_dir = Path(src_dataset_dir)\n    dst_dataset_dir = Path(dst_dataset_dir)\n\n    if dst_dataset_dir.exists():\n        shutil.rmtree(dst_dataset_dir)\n    dst_dataset_dir.mkdir(parents=True, exist_ok=True)\n\n    # Copy valid/test as-is\n    for split in [\"valid\", \"test\"]:\n        shutil.copytree(src_dataset_dir / split, dst_dataset_dir / split, dirs_exist_ok=True)\n\n    src_train = src_dataset_dir / \"train\"\n    dst_train = dst_dataset_dir / \"train\"\n    dst_train.mkdir(parents=True, exist_ok=True)\n\n    coco_path = src_train / \"_annotations.coco.json\"\n    coco = json.loads(coco_path.read_text())\n\n    images = coco[\"images\"]\n    anns   = coco[\"annotations\"]\n    cats   = coco.get(\"categories\", [])\n\n    ann_by_img = {}\n    for a in anns:\n        ann_by_img.setdefault(a[\"image_id\"], []).append(a)\n\n    # Copy originals\n    for im in images:\n        _safe_symlink(src_train / im[\"file_name\"], dst_train / im[\"file_name\"])\n\n    new_images = [dict(im) for im in images]\n    new_anns   = [dict(a) for a in anns]\n    next_img_id = max([im[\"id\"] for im in images] + [0]) + 1\n    next_ann_id = max([a[\"id\"] for a in anns] + [0]) + 1\n\n    n_base = len(images)\n    n_aug  = int(round(n_base * float(aug_ratio)))\n    n_mix  = int(round(n_base * float(mixup_ratio)))\n\n    print(f\"Train images: {n_base}\")\n    print(f\"Generating aug samples:   {n_aug}\")\n    print(f\"Generating mixup samples: {n_mix}\")\n\n    # Aug samples\n    for i in range(n_aug):\n        base_im = images[np.random.randint(n_base)]\n        base_path = src_train / base_im[\"file_name\"]\n        base_anns = ann_by_img.get(base_im[\"id\"], [])\n        b_in = [a[\"bbox\"] for a in base_anns]\n        c_in = [a[\"category_id\"] for a in base_anns]\n\n        out_name = f\"{Path(base_im['file_name']).stem}__aug_{i:06d}{Path(base_im['file_name']).suffix.lower()}\"\n        out_path = dst_train / out_name\n\n        pipeline = AUG_GEOM if (USE_GEOM and (np.random.rand() < GEOM_PROB)) else AUG_COLOR\n        b_out, c_out = _apply_aug(base_path, b_in, c_in, out_path, pipeline)\n        if b_out is None:\n            continue\n\n        new_im = dict(base_im)\n        new_im[\"id\"] = next_img_id\n        new_im[\"file_name\"] = out_name\n        new_images.append(new_im)\n\n        for bb, cid in zip(b_out, c_out):\n            new_anns.append({\n                \"id\": next_ann_id,\n                \"image_id\": next_img_id,\n                \"category_id\": int(cid),\n                \"bbox\": [float(x) for x in bb],\n                \"area\": float(bb[2] * bb[3]),\n                \"iscrowd\": 0,\n            })\n            next_ann_id += 1\n        next_img_id += 1\n\n        if (i+1) % 500 == 0:\n            print(f\"  aug progress: {i+1}/{n_aug}\")\n\n    # MixUp samples\n    for i in range(n_mix):\n        imA = images[np.random.randint(n_base)]\n        imB = images[np.random.randint(n_base)]\n        pathA = src_train / imA[\"file_name\"]\n        pathB = src_train / imB[\"file_name\"]\n\n        imgA = cv2.imread(str(pathA))\n        imgB = cv2.imread(str(pathB))\n        if imgA is None or imgB is None:\n            continue\n        if imgA.shape[:2] != imgB.shape[:2]:\n            imgB = cv2.resize(imgB, (imgA.shape[1], imgA.shape[0]))\n\n        annsA = ann_by_img.get(imA[\"id\"], [])\n        annsB = ann_by_img.get(imB[\"id\"], [])\n        bA = [a[\"bbox\"] for a in annsA]; cA = [a[\"category_id\"] for a in annsA]\n        bB = [a[\"bbox\"] for a in annsB]; cB = [a[\"category_id\"] for a in annsB]\n\n        mixed, b_out, c_out = _mixup(imgA, bA, cA, imgB, bB, cB)\n\n        out_name = f\"{Path(imA['file_name']).stem}__mixup_{i:06d}{Path(imA['file_name']).suffix.lower()}\"\n        out_path = dst_train / out_name\n        cv2.imwrite(str(out_path), mixed)\n\n        new_im = dict(imA)\n        new_im[\"id\"] = next_img_id\n        new_im[\"file_name\"] = out_name\n        new_images.append(new_im)\n\n        for bb, cid in zip(b_out, c_out):\n            new_anns.append({\n                \"id\": next_ann_id,\n                \"image_id\": next_img_id,\n                \"category_id\": int(cid),\n                \"bbox\": [float(x) for x in bb],\n                \"area\": float(bb[2] * bb[3]),\n                \"iscrowd\": 0,\n            })\n            next_ann_id += 1\n        next_img_id += 1\n\n        if (i+1) % 250 == 0:\n            print(f\"  mixup progress: {i+1}/{n_mix}\")\n\n    out_coco = dict(coco)\n    out_coco[\"images\"] = new_images\n    out_coco[\"annotations\"] = new_anns\n    out_coco[\"categories\"] = cats\n\n    (dst_train / \"_annotations.coco.json\").write_text(json.dumps(out_coco))\n    print(f\"Augmented train: images={len(new_images)} annotations={len(new_anns)}\")\n    return dst_dataset_dir\n\nif AUG_RATIO > 0.0 or MIXUP_RATIO > 0.0:\n    DATASET_USED = augment_train_coco(COCO_DATASET_DIR, AUG_DATASET_DIR, aug_ratio=AUG_RATIO, mixup_ratio=MIXUP_RATIO)\nelse:\n    # no augmentation => just use the base dataset\n    DATASET_USED = COCO_DATASET_DIR\n\nprint(\"DATASET_USED:\", DATASET_USED)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# =========================\n# 4) Build two single-class datasets (FACE-only, SHEMAGH-only) from the SAME DATASET_USED\n# =========================\ndef build_single_class_dataset(src_dataset_dir: Path, dst_dataset_dir: Path, keep_coco_cat_id: int, new_name: str):\n    '''\n    Create a dataset dir with the same splits/images, but annotations filtered to ONE category.\n\n    Also robust to occasional dataset inconsistencies (COCO json referencing a missing image file)\n    by filtering out such images BEFORE training.\n    '''\n    src_dataset_dir = Path(src_dataset_dir)\n    dst_dataset_dir = Path(dst_dataset_dir)\n    if dst_dataset_dir.exists():\n        shutil.rmtree(dst_dataset_dir)\n    dst_dataset_dir.mkdir(parents=True, exist_ok=True)\n\n    for split in [\"train\", \"valid\", \"test\"]:\n        src_split = src_dataset_dir / split\n        dst_split = dst_dataset_dir / split\n        dst_split.mkdir(parents=True, exist_ok=True)\n\n        coco_path = src_split / \"_annotations.coco.json\"\n        coco = json.loads(coco_path.read_text())\n\n        disk_files = {p.name for p in _list_images(src_split)}\n\n        images_filtered = []\n        valid_img_ids = set()\n        for im in coco.get(\"images\", []):\n            fn = Path(im.get(\"file_name\", \"\")).name\n            if fn in disk_files:\n                im2 = dict(im)\n                im2[\"file_name\"] = fn\n                images_filtered.append(im2)\n                valid_img_ids.add(int(im2[\"id\"]))\n\n        for im in images_filtered:\n            fn = im[\"file_name\"]\n            _safe_symlink(src_split / fn, dst_split / fn, prefer_symlink=True)\n\n        anns = []\n        ann_id = 1\n        for a in coco.get(\"annotations\", []):\n            if int(a.get(\"image_id\", -1)) not in valid_img_ids:\n                continue\n            if int(a.get(\"category_id\", -1)) != int(keep_coco_cat_id):\n                continue\n            a2 = dict(a)\n            a2[\"id\"] = ann_id\n            a2[\"category_id\"] = 1\n            anns.append(a2)\n            ann_id += 1\n\n        out = dict(coco)\n        out[\"images\"] = images_filtered\n        out[\"annotations\"] = anns\n        out[\"categories\"] = [{\"id\": 1, \"name\": new_name, \"supercategory\": \"object\"}]\n\n        (dst_split / \"_annotations.coco.json\").write_text(json.dumps(out))\n\n        _assert_coco_split_has_files(dst_split)\n        print(f\"[{dst_dataset_dir.name}/{split}] images={len(images_filtered)} anns={len(anns)}\")\n\n    return dst_dataset_dir\n\nFACE_DATASET = build_single_class_dataset(DATASET_USED, FACE_DATASET_DIR, keep_coco_cat_id=1, new_name=\"face\")\nSHEM_DATASET = build_single_class_dataset(DATASET_USED, SHEM_DATASET_DIR, keep_coco_cat_id=2, new_name=\"shemagh\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Phase 3 — Utilities (training + inference + postprocess + evaluation + visualization)","metadata":{}},{"cell_type":"code","source":"# =========================\n# 5) Train RF-DETR (configurable)\n# =========================\n# Optional HF login (only needed if pretrained weights are gated)\ntry:\n    from huggingface_hub import login\n    hf_token = os.getenv(\"HF_TOKEN\")\n    if hf_token:\n        login(token=hf_token, add_to_git_credential=False)\nexcept Exception:\n    pass\n\n# ---- Model factory ----\ndef _normalize_model_name(name: str) -> str:\n    name = (name or \"small\").lower().strip()\n    name = name.replace(\"rf-detr-\", \"\").replace(\"rfdetr-\", \"\").replace(\"rf_detr_\", \"\")\n    return name\n\ndef get_rfdetr_model_class(model_name: str):\n    name = _normalize_model_name(model_name)\n    mapping = {\n        \"small\": RFDETRSmall,\n        \"medium\": RFDETRMedium,\n        \"large\": RFDETRLarge,\n        \"base\": RFDETRBase,\n    }\n    if name not in mapping:\n        raise ValueError(f\"Unknown model_name={model_name!r}. Use one of: {sorted(mapping.keys())}\")\n    return mapping[name]\n\ndef _manual_load_weights(wrapper_model, ckpt_path: Path):\n    ckpt = torch.load(str(ckpt_path), map_location=\"cpu\")\n    if isinstance(ckpt, dict):\n        if \"model\" in ckpt and isinstance(ckpt[\"model\"], dict):\n            state = ckpt[\"model\"]\n        elif \"state_dict\" in ckpt and isinstance(ckpt[\"state_dict\"], dict):\n            state = ckpt[\"state_dict\"]\n        else:\n            state = ckpt\n    else:\n        state = ckpt\n\n    candidates = [wrapper_model]\n    for attr in [\"model\", \"_model\"]:\n        if hasattr(wrapper_model, attr):\n            candidates.append(getattr(wrapper_model, attr))\n    for c in list( candidates ):\n        if c is None:\n            continue\n        if hasattr(c, \"model\"):\n            candidates.append(getattr(c, \"model\"))\n    for c in candidates:\n        if c is None:\n            continue\n        if hasattr(c, \"load_state_dict\"):\n            try:\n                c.load_state_dict(state, strict=False)\n                return\n            except Exception:\n                pass\n    raise RuntimeError(f\"Failed to load checkpoint weights manually from: {ckpt_path}\")\n\ndef make_rfdetr(\n    model_name: str = \"small\",\n    *,\n    resolution: int = 224,\n    num_classes: Optional[int] = None,\n    class_names: Optional[List[str]] = None,\n    pretrain_weights: Optional[Path] = None,\n    optimize: bool = False,\n    optimize_compile: bool = True,\n    optimize_batch_size: int = 1,\n):\n    '''\n    Create an RF-DETR model wrapper.\n\n    Note: rfdetr does NOT expose a generic `.load()` on the wrapper.\n    To load weights, pass `pretrain_weights` (or `pretrained_weights` on some versions)\n    at construction time; otherwise we fall back to a manual state_dict load.\n    '''\n    ModelCls = get_rfdetr_model_class(model_name)\n    sig = inspect.signature(ModelCls.__init__)\n\n    kwargs: Dict[str, Any] = {\"resolution\": int(resolution)}\n    if num_classes is not None and \"num_classes\" in sig.parameters:\n        kwargs[\"num_classes\"] = int(num_classes)\n    if class_names is not None and \"class_names\" in sig.parameters:\n        kwargs[\"class_names\"] = list(class_names)\n\n    passed_pretrain = False\n    if pretrain_weights is not None:\n        ckpt = str(Path(pretrain_weights))\n        if \"pretrain_weights\" in sig.parameters:\n            kwargs[\"pretrain_weights\"] = ckpt\n            passed_pretrain = True\n        elif \"pretrained_weights\" in sig.parameters:\n            kwargs[\"pretrained_weights\"] = ckpt\n            passed_pretrain = True\n        elif \"pretrained_weight\" in sig.parameters:\n            kwargs[\"pretrained_weight\"] = ckpt\n            passed_pretrain = True\n\n    model = ModelCls(**kwargs)\n\n    if (pretrain_weights is not None) and (not passed_pretrain):\n        _manual_load_weights(model, Path(pretrain_weights))\n\n    if optimize:\n        try:\n            model.optimize_for_inference(batch_size=int(optimize_batch_size), compile=bool(optimize_compile))\n        except Exception:\n            try:\n                model.optimize_for_inference(batch_size=int(optimize_batch_size), compile=False)\n            except Exception:\n                pass\n\n    return model\n\n# ---- Training pipeline ----\n@dataclass\nclass TrainParams:\n    model_name: str = \"small\"\n    resolution: int = MODEL_RESOLUTION\n    epochs: int = EPOCHS\n    batch_size: int = BATCH_SIZE\n    grad_accum: int = GRAD_ACCUM\n    lr: float = LR\n    device: Optional[str] = None\n    num_workers: int = 4\n    extra_train_kwargs: Optional[Dict[str, Any]] = None\n\ndef train_one(dataset_dir: Path, out_dir: Path, params: TrainParams):\n    set_seed(SEED)\n    out_dir = Path(out_dir)\n    out_dir.mkdir(parents=True, exist_ok=True)\n\n    if params.device is None:\n        params.device = \"cuda\" if torch.cuda.is_available() else \"cpu\"\n\n    init_w = globals().get(\"INIT_PRETRAIN_WEIGHTS\", None)\n    # If the file is not found, init_w stays None and training starts from scratch.\n    model = make_rfdetr(params.model_name, resolution=params.resolution, pretrain_weights=init_w)\n\n    extra = dict(\n        gradient_checkpointing=True,\n        print_freq=5000,\n        threshold=0.5,\n        tensorboard=False,\n        run_test=False,\n    )\n    if params.extra_train_kwargs:\n        extra.update(params.extra_train_kwargs)\n\n    model.train(\n        dataset_dir=str(dataset_dir),\n        epochs=int(params.epochs),\n        batch_size=int(params.batch_size),\n        grad_accum_steps=int(params.grad_accum),\n        lr=float(params.lr),\n        lr_drop = 5,\n        output_dir=str(out_dir),\n        device=str(params.device),\n        num_workers=int(params.num_workers),\n        **extra,\n    )\n    return model, out_dir","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# =========================\n# Inference & postprocess utilities (robust checkpoint loading)\n# =========================\ndef load_rfdetr_for_inference(\n    ckpt_path: Path,\n    *,\n    model_name: str = \"small\",\n    resolution: int = 224,\n    optimize: bool = False,\n    optimize_compile: bool = False,\n    optimize_batch_size: int = 1,\n):\n    \"\"\"Load an RF-DETR model from a training checkpoint.\n\n    IMPORTANT: Follow upstream usage: ModelCls(pretrain_weights=\"/path/to/checkpoint.pth\")\n    and allow-list common pickled globals for PyTorch >= 2.6 weights-only loading.\n    \"\"\"\n    ckpt_path = Path(ckpt_path)\n    if not ckpt_path.exists():\n        raise FileNotFoundError(f\"Checkpoint not found: {ckpt_path}\")\n\n    ModelCls = get_rfdetr_model_class(model_name)\n    sig = inspect.signature(ModelCls.__init__)\n\n    kwargs: Dict[str, Any] = {\"resolution\": int(resolution)}\n    # rfdetr versions vary in the kwarg name\n    for k in (\"pretrain_weights\", \"pretrained_weights\", \"pretrained_weight\"):\n        if k in sig.parameters:\n            kwargs[k] = str(ckpt_path)\n            break\n    else:\n        raise RuntimeError(\"Your rfdetr version does not accept pretrain_weights/pretrained_weights.\")\n\n    # PyTorch>=2.6 default weights_only=True can require allow-listing certain pickled globals.\n    allow = [argparse.Namespace]\n    try:\n        from pathlib import PosixPath, WindowsPath\n        allow += [PosixPath, WindowsPath]\n    except Exception:\n        pass\n\n    with safe_globals(allow):\n        model = ModelCls(**kwargs)\n\n    if optimize:\n        try:\n            model.optimize_for_inference(batch_size=int(optimize_batch_size), compile=bool(optimize_compile))\n        except Exception:\n            try:\n                model.optimize_for_inference(batch_size=int(optimize_batch_size), compile=False)\n            except Exception:\n                pass\n\n    return model\n\n\ndef _xyxy_to_yolo_norm_str(xyxy, w, h):\n    x1, y1, x2, y2 = [float(v) for v in xyxy]\n    x1 = max(0.0, min(x1, w))\n    x2 = max(0.0, min(x2, w))\n    y1 = max(0.0, min(y1, h))\n    y2 = max(0.0, min(y2, h))\n    bw = max(0.0, x2 - x1)\n    bh = max(0.0, y2 - y1)\n    xc = x1 + bw / 2.0\n    yc = y1 + bh / 2.0\n    # normalized YOLO\n    return f\"{xc / w:.6f} {yc / h:.6f} {bw / w:.6f} {bh / h:.6f}\"\n\n\ndef detections_to_yolo_predstring(det, w, h, fixed_yolo_cls: int):\n    # det is expected to be supervision.Detections (returned by model.predict)\n    if det is None:\n        return \"\"\n    xyxy = getattr(det, \"xyxy\", None)\n    conf = getattr(det, \"confidence\", None)\n    if xyxy is None or conf is None:\n        return \"\"\n    xyxy = np.asarray(xyxy)\n    conf = np.asarray(conf)\n\n    parts = []\n    for box, score in zip(xyxy, conf):\n        parts.append(str(int(fixed_yolo_cls)))\n        parts.append(f\"{float(score):.6f}\")\n        parts.append(_xyxy_to_yolo_norm_str(box, w, h))\n    return \" \".join(parts)\n\n\ndef predict_folder(\n    model,\n    img_dir: Path,\n    *,\n    conf_th: float,\n    fixed_yolo_cls: int,\n    max_images: Optional[int] = None,\n    save_csv: Optional[Path] = None,\n    tta: bool = True,\n    nms_iou: float = 0.65,\n):\n    '''\n    Run RF-DETR inference on a folder and return a DF with (filename, prediction_string).\n\n    Notes:\n    - `conf_th` is passed into `model.predict(..., threshold=conf_th)` (low thresholds keep more boxes).\n    - If `tta=True`, we also run a horizontal-flip pass and merge with a simple NMS.\n    '''\n    img_dir = Path(img_dir)\n    paths = sorted([p for p in img_dir.glob(\"*.jpg\")]) + sorted([p for p in img_dir.glob(\"*.png\")])\n    if max_images is not None:\n        paths = paths[: int(max_images)]\n\n    def _bbox_iou_xyxy(a, b):\n        ax1, ay1, ax2, ay2 = [float(x) for x in a]\n        bx1, by1, bx2, by2 = [float(x) for x in b]\n        ix1, iy1 = max(ax1, bx1), max(ay1, by1)\n        ix2, iy2 = min(ax2, bx2), min(ay2, by2)\n        iw, ih = max(0.0, ix2 - ix1), max(0.0, iy2 - iy1)\n        inter = iw * ih\n        area_a = max(0.0, ax2 - ax1) * max(0.0, ay2 - ay1)\n        area_b = max(0.0, bx2 - bx1) * max(0.0, by2 - by1)\n        denom = area_a + area_b - inter\n        return (inter / denom) if denom > 0 else 0.0\n\n    def _nms_xyxy(boxes, scores, iou_th):\n        if len(boxes) == 0:\n            return []\n        boxes = np.asarray(boxes, dtype=float)\n        scores = np.asarray(scores, dtype=float)\n        order = scores.argsort()[::-1]\n        keep = []\n        while order.size > 0:\n            i = int(order[0])\n            keep.append(i)\n            if order.size == 1:\n                break\n            rest = order[1:]\n            ious = np.array([_bbox_iou_xyxy(boxes[i], boxes[j]) for j in rest], dtype=float)\n            rest = rest[ious <= float(iou_th)]\n            order = rest\n        return keep\n\n    def _det_to_arrays(det):\n        if det is None:\n            return np.zeros((0, 4), dtype=float), np.zeros((0,), dtype=float)\n        xyxy = getattr(det, \"xyxy\", None)\n        conf = getattr(det, \"confidence\", None)\n        if xyxy is None or conf is None:\n            return np.zeros((0, 4), dtype=float), np.zeros((0,), dtype=float)\n        xyxy = np.asarray(xyxy, dtype=float).reshape(-1, 4)\n        conf = np.asarray(conf, dtype=float).reshape(-1)\n        if len(xyxy) != len(conf):\n            n = min(len(xyxy), len(conf))\n            xyxy = xyxy[:n]\n            conf = conf[:n]\n        return xyxy, conf\n\n    def _predict_any(inp):\n        # Prefer passing a path (upstream examples). Fallback to array if needed.\n        try:\n            return model.predict(str(inp), threshold=float(conf_th))\n        except Exception:\n            return model.predict(inp, threshold=float(conf_th))\n\n    rows = []\n    for p in paths:\n        pred = \"\"\n        try:\n            im = Image.open(p).convert(\"RGB\")\n            w, h = im.size\n\n            det = _predict_any(p)\n            boxes, scores = _det_to_arrays(det)\n\n            if bool(tta):\n                # Horizontal flip TTA\n                arr = np.array(im)  # RGB\n                arr_f = np.ascontiguousarray(arr[:, ::-1, :])\n                try:\n                    det_f = _predict_any(arr_f)\n                    b_f, s_f = _det_to_arrays(det_f)\n                except Exception:\n                    b_f, s_f = np.zeros((0, 4), dtype=float), np.zeros((0,), dtype=float)\n\n                if len(b_f) > 0:\n                    # map flipped coords back\n                    # x' = W - x  (for both x1/x2, then swap)\n                    x1 = float(w) - b_f[:, 2]\n                    x2 = float(w) - b_f[:, 0]\n                    b_f = b_f.copy()\n                    b_f[:, 0] = x1\n                    b_f[:, 2] = x2\n\n                    boxes = np.vstack([boxes, b_f]) if len(boxes) else b_f\n                    scores = np.concatenate([scores, s_f]) if len(scores) else s_f\n\n            # NMS on merged boxes\n            keep = _nms_xyxy(boxes, scores, float(nms_iou)) if (len(boxes) > 0) else []\n            parts = []\n            for i in keep:\n                box = boxes[i]\n                score = scores[i]\n                parts.append(str(int(fixed_yolo_cls)))\n                parts.append(f\"{float(score):.6f}\")\n                parts.append(_xyxy_to_yolo_norm_str(box, w, h))\n            pred = \" \".join(parts)\n\n        except Exception as e:\n            print(\"WARN: failed on\", p.name, \"|\", repr(e))\n            pred = \"\"\n\n        rows.append({\"filename\": p.name, \"prediction_string\": pred})\n\n    df = pd.DataFrame(rows)\n    if save_csv is not None:\n        save_csv = Path(save_csv)\n        df.to_csv(save_csv, index=False)\n        print(\"Wrote:\", save_csv, \"(rows=\", len(df), \")\")\n    return df\ndef combine_preds(df_a: pd.DataFrame, df_b: pd.DataFrame) -> pd.DataFrame:\n    \"\"\"Outer-join on filename and concatenate prediction strings.\"\"\"\n    a = df_a.copy()\n    b = df_b.copy()\n    a = a.rename(columns={\"prediction_string\": \"pred_a\"})\n    b = b.rename(columns={\"prediction_string\": \"pred_b\"})\n    m = a.merge(b, on=\"filename\", how=\"outer\")\n    m[\"pred_a\"] = m[\"pred_a\"].fillna(\"\")\n    m[\"pred_b\"] = m[\"pred_b\"].fillna(\"\")\n    m[\"prediction_string\"] = (m[\"pred_a\"].str.strip() + \" \" + m[\"pred_b\"].str.strip()).str.strip()\n    return m[[\"filename\", \"prediction_string\"]]\n\n\n# =========================\n# Right-place utilities (containment-based)\n# =========================\ndef parse_predstring(s: str):\n    \"\"\"Parse 'cls conf xc yc w h ...' into lists.\"\"\"\n    s = (s or \"\").strip()\n    if not s:\n        return [], [], []\n    toks = s.split()\n    if len(toks) % 6 != 0:\n        # malformed; best-effort: drop remainder\n        toks = toks[: len(toks) - (len(toks) % 6)]\n    cls_ids, confs, boxes = [], [], []\n    for i in range(0, len(toks), 6):\n        cls_ids.append(int(float(toks[i])))\n        confs.append(float(toks[i + 1]))\n        xc, yc, bw, bh = map(float, toks[i + 2 : i + 6])\n        boxes.append((xc, yc, bw, bh))\n    return cls_ids, confs, boxes\n\n\ndef norm_xywh_to_xyxy(xc, yc, bw, bh):\n    x1 = xc - bw / 2.0\n    y1 = yc - bh / 2.0\n    x2 = xc + bw / 2.0\n    y2 = yc + bh / 2.0\n    return x1, y1, x2, y2\n\n\ndef bbox_iou(a, b):\n    ax1, ay1, ax2, ay2 = a\n    bx1, by1, bx2, by2 = b\n    ix1, iy1 = max(ax1, bx1), max(ay1, by1)\n    ix2, iy2 = min(ax2, bx2), min(ay2, by2)\n    iw, ih = max(0.0, ix2 - ix1), max(0.0, iy2 - iy1)\n    inter = iw * ih\n    area_a = max(0.0, ax2 - ax1) * max(0.0, ay2 - ay1)\n    area_b = max(0.0, bx2 - bx1) * max(0.0, by2 - by1)\n    denom = area_a + area_b - inter\n    return (inter / denom) if denom > 0 else 0.0\n\n\ndef contain_ratio(inner, outer):\n    \"\"\"Return fraction of 'inner' area contained within 'outer'.\"\"\"\n    ix1, iy1, ix2, iy2 = inner\n    ox1, oy1, ox2, oy2 = outer\n    cx1, cy1 = max(ix1, ox1), max(iy1, oy1)\n    cx2, cy2 = min(ix2, ox2), min(iy2, oy2)\n    iw, ih = max(0.0, cx2 - cx1), max(0.0, cy2 - cy1)\n    inter = iw * ih\n    area_inner = max(0.0, ix2 - ix1) * max(0.0, iy2 - iy1)\n    return (inter / area_inner) if area_inner > 0 else 0.0\n\n\ndef pred_right_place_from_predstring(pred_string: str, *, face_cls: int, shem_cls: int, contain_th: float):\n    cls_ids, confs, boxes = parse_predstring(pred_string)\n    if not cls_ids:\n        return 0\n\n\n\ndef pred_right_place_from_predstring_confident(\n    pred_string: str,\n    *,\n    face_cls: int,\n    shem_cls: int,\n    contain_th: float,\n    certain_th: float,\n):\n    '''\n    Decide right_place from predictions with an additional \"certainty\" threshold:\n    - Only consider FACE/SHEMAGH boxes with confidence >= certain_th.\n    - Then apply the containment threshold contain_th.\n    '''\n    cls_ids, confs, boxes = parse_predstring(pred_string)\n    if not cls_ids:\n        return 0\n\n    face_xyxys = []\n    shem_xyxys = []\n\n    for cid, score, box in zip(cls_ids, confs, boxes):\n        if float(score) < float(certain_th):\n            continue\n        x1, y1, x2, y2 = norm_xywh_to_xyxy(*box)\n        if cid == face_cls:\n            face_xyxys.append((x1, y1, x2, y2))\n        elif cid == shem_cls:\n            shem_xyxys.append((x1, y1, x2, y2))\n\n    if not face_xyxys or not shem_xyxys:\n        return 0\n\n    for f in face_xyxys:\n        for s in shem_xyxys:\n            if contain_ratio(f, s) >= float(contain_th):\n                return 1\n    return 0\n    face_xyxys = []\n    shem_xyxys = []\n\n    for cid, (_, box) in zip(cls_ids, zip(confs, boxes)):\n        x1, y1, x2, y2 = norm_xywh_to_xyxy(*box)\n        if cid == face_cls:\n            face_xyxys.append((x1, y1, x2, y2))\n        elif cid == shem_cls:\n            shem_xyxys.append((x1, y1, x2, y2))\n\n    if not face_xyxys or not shem_xyxys:\n        return 0\n\n    # 'right_place' if ANY face is sufficiently contained in ANY shemagh\n    for f in face_xyxys:\n        for s in shem_xyxys:\n            if contain_ratio(f, s) >= float(contain_th):\n                return 1\n    return 0\n\n\ndef f1_score_binary(y_true, y_pred):\n    y_true = np.asarray(y_true).astype(int)\n    y_pred = np.asarray(y_pred).astype(int)\n    tp = int(((y_true == 1) & (y_pred == 1)).sum())\n    fp = int(((y_true == 0) & (y_pred == 1)).sum())\n    fn = int(((y_true == 1) & (y_pred == 0)).sum())\n    denom = 2 * tp + fp + fn\n    return (2 * tp / denom) if denom > 0 else 0.0\n\n\n# =========================\n# Visualization helper\n# =========================\ndef plot_image_with_boxes(img_path: Path, pred_string: str, *, title: str = \"\", max_boxes: int = 50):\n    img_path = Path(img_path)\n    im = Image.open(img_path).convert(\"RGB\")\n    w, h = im.size\n    fig, ax = plt.subplots(figsize=(5, 5))\n    ax.imshow(im)\n    ax.set_title(title)\n    ax.axis(\"off\")\n\n    cls_ids, confs, boxes = parse_predstring(pred_string)\n    for idx, (cid, score, box) in enumerate(zip(cls_ids, confs, boxes)):\n        if idx >= max_boxes:\n            break\n        xc, yc, bw, bh = box\n        # denormalize\n        x1 = (xc - bw / 2.0) * w\n        y1 = (yc - bh / 2.0) * h\n        x2 = (xc + bw / 2.0) * w\n        y2 = (yc + bh / 2.0) * h\n        rect = patches.Rectangle((x1, y1), x2 - x1, y2 - y1, fill=False, linewidth=2)\n        ax.add_patch(rect)\n        ax.text(x1, y1, f\"{cid}:{score:.2f}\", fontsize=10, bbox=dict(facecolor=\"white\", alpha=0.6))\n    plt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Phase 4 — Train + infer Face model","metadata":{}},{"cell_type":"code","source":"# =========================\n# Phase 4) Train FACE model (keep the trained model in-memory)\n# =========================\n\n# Note:\n# - This trains the model and keeps `model_face` in RAM so you can run inference immediately\n#   without reloading any checkpoints (avoids class-head mismatch issues).\n# - A checkpoint is still saved to disk by RF-DETR for reproducibility.\n\nFACE_TRAIN = TrainParams(\n    model_name=\"medium\",\n    resolution=MODEL_RESOLUTION,\n    epochs=EPOCHS,\n    batch_size=BATCH_SIZE,\n    grad_accum=GRAD_ACCUM,\n    lr=LR,\n)\n\nif \"model_face\" not in globals():\n    model_face, _ = train_one(FACE_DATASET, OUT_FACE, FACE_TRAIN)\nelse:\n    print(\"Using existing model_face from memory (skipping training).\")\n\nCKPT_FACE = OUT_FACE / \"checkpoint_best_total.pth\"\nprint(\"FACE ckpt:\", CKPT_FACE)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# =========================\n# Phase 5) Inference FACE on train/valid/test (NO checkpoint reload)\n# =========================\n\nassert \"model_face\" in globals(), \"model_face is not defined. Run the FACE training cell first.\"\n\nTRAIN_DIR = COCO_DATASET_DIR / \"train\"\nVAL_DIR   = COCO_DATASET_DIR / \"valid\"\n\n\n# Put the trained model into inference mode (best-effort, compatible across rfdetr versions).\ntry:\n    model_face.optimize_for_inference(batch_size=1, compile=False)\nexcept Exception:\n    pass\n\nface_train_df = predict_folder(\n    model_face, TRAIN_DIR,\n    conf_th=CONF_FACE,\n    fixed_yolo_cls=FACE_YOLO,\n    save_csv=Path(\"preds_face_train.csv\"),\n    tta=USE_TTA,\n    nms_iou=NMS_IOU,\n)\nface_val_df = predict_folder(\n    model_face, VAL_DIR,\n    conf_th=CONF_FACE,\n    fixed_yolo_cls=FACE_YOLO,\n    save_csv=Path(\"preds_face_valid.csv\"),\n    tta=USE_TTA,\n    nms_iou=NMS_IOU,\n)\nface_test_df = predict_folder(\n    model_face, TEST_IMAGES_DIR,\n    conf_th=CONF_FACE,\n    fixed_yolo_cls=FACE_YOLO,\n    save_csv=Path(\"preds_face_test.csv\"),\n    tta=USE_TTA,\n    nms_iou=NMS_IOU,\n)\n\ndisplay(face_test_df.head())","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Phase 5 — Train + infer Shemagh model","metadata":{}},{"cell_type":"code","source":"# =========================\n# Phase 6) Train SHEMAGH model (keep the trained model in-memory)\n# =========================\n\nSHEM_TRAIN = TrainParams(\n    model_name=\"medium\",\n    resolution=MODEL_RESOLUTION,\n    epochs=EPOCHS,\n    batch_size=BATCH_SIZE,\n    grad_accum=GRAD_ACCUM,\n    lr=LR\n)\n\nif \"model_shemagh\" not in globals():\n    model_shemagh, _ = train_one(SHEM_DATASET, OUT_SHEM, SHEM_TRAIN)\nelse:\n    print(\"Using existing model_shemagh from memory (skipping training).\")\n\nCKPT_SHEM = OUT_SHEM / \"checkpoint_best_total.pth\"\nprint(\"SHEMAGH ckpt:\", CKPT_SHEM)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# =========================\n# Phase 7) Inference SHEMAGH on train/valid/test (NO checkpoint reload)\n# =========================\n\nassert \"model_shemagh\" in globals(), \"model_shemagh is not defined. Run the SHEMAGH training cell first.\"\n\nTRAIN_DIR = COCO_DATASET_DIR / \"train\"\nVAL_DIR   = COCO_DATASET_DIR / \"valid\"\n\n\ntry:\n    model_shemagh.optimize_for_inference(batch_size=1, compile=False)\nexcept Exception:\n    pass\n\nshem_train_df = predict_folder(\n    model_shemagh, TRAIN_DIR,\n    conf_th=CONF_SHEM,\n    fixed_yolo_cls=SHEM_YOLO,\n    save_csv=Path(\"preds_shem_train.csv\"),\n    tta=USE_TTA,\n    nms_iou=NMS_IOU,\n)\nshem_val_df = predict_folder(\n    model_shemagh, VAL_DIR,\n    conf_th=CONF_SHEM,\n    fixed_yolo_cls=SHEM_YOLO,\n    save_csv=Path(\"preds_shem_valid.csv\"),\n    tta=USE_TTA,\n    nms_iou=NMS_IOU,\n)\nshem_test_df = predict_folder(\n    model_shemagh, TEST_IMAGES_DIR,\n    conf_th=CONF_SHEM,\n    fixed_yolo_cls=SHEM_YOLO,\n    save_csv=Path(\"preds_shem_test.csv\"),\n    tta=USE_TTA,\n    nms_iou=NMS_IOU,\n)\n\ndisplay(shem_test_df.head())","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Phase 6 — Combine predictions + tune containment threshold (right_place)","metadata":{}},{"cell_type":"code","source":"# Combine FACE + SHEMAGH predictions for each split\ntrain_combined = combine_preds(face_train_df, shem_train_df)\nval_combined   = combine_preds(face_val_df, shem_val_df)\ntest_combined  = combine_preds(face_test_df, shem_test_df)\ntest_combined.to_csv(\"combined_test_df.csv\",index = False)\n\nprint(\"combined rows:\", len(test_combined))\ndisplay(test_combined.head())\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# =========================\n# Phase 8) Threshold tuning (as intended)\n#   B) Use GT boxes + TRAIN split to find BEST_CONTAIN_TH\n#   C) Use predicted VAL boxes + a new threshold (BEST_CERTAIN_TH) to decide \"confident\" boxes\n#      Then compute VALID score = (mAP50:95 + F1)/2\n# =========================\n\nfrom pycocotools.coco import COCO\nfrom pycocotools.cocoeval import COCOeval\n\nTRAIN_SPLIT_CSV = YOLO_SPLIT_DIR / \"train_labels_split.csv\"\nVAL_SPLIT_CSV   = YOLO_SPLIT_DIR / \"val_labels_split.csv\"\n\ntrain_labels = pd.read_csv(TRAIN_SPLIT_CSV)[[\"filename\", \"right_place\"]].copy()\nval_labels   = pd.read_csv(VAL_SPLIT_CSV)[[\"filename\", \"right_place\"]].copy()\n\ntrain_labels[\"right_place\"] = train_labels[\"right_place\"].fillna(0).astype(int)\nval_labels[\"right_place\"]   = val_labels[\"right_place\"].fillna(0).astype(int)\n\n# ------------------------------------------------------------\n# B) Tune containment threshold on TRAIN using GT boxes\n# ------------------------------------------------------------\nTRAIN_GT_LABELS_DIR = YOLO_SPLIT_DIR / \"labels\" / \"train\"\n\ndef _max_gt_contain_ratio(filename: str) -> float:\n    # Return max(face inside shemagh) ratio using GT boxes for this image.\n    # If missing face/shem pair -> return -1 so it never predicts positive (even at th=0).\n    stem = Path(filename).stem\n    lab_path = TRAIN_GT_LABELS_DIR / f\"{stem}.txt\"\n    if not lab_path.exists():\n        return -1.0\n\n    face_xyxys = []\n    shem_xyxys = []\n\n    for line in lab_path.read_text(encoding=\"utf-8\").splitlines():\n        line = line.strip()\n        if not line:\n            continue\n        t = line.split()\n        if len(t) < 5:\n            continue\n        cid = int(float(t[0]))\n        xc, yc, bw, bh = map(float, t[1:5])\n        x1, y1, x2, y2 = norm_xywh_to_xyxy(xc, yc, bw, bh)\n        if cid == FACE_YOLO:\n            face_xyxys.append((x1, y1, x2, y2))\n        elif cid == SHEM_YOLO:\n            shem_xyxys.append((x1, y1, x2, y2))\n\n    if (not face_xyxys) or (not shem_xyxys):\n        return -1.0\n\n    best = 0.0\n    for f in face_xyxys:\n        for s in shem_xyxys:\n            best = max(best, float(contain_ratio(f, s)))\n    return float(best)\n\ntrain_max_ratios = np.array([_max_gt_contain_ratio(fn) for fn in train_labels[\"filename\"].tolist()], dtype=float)\n\nbest_contain_th = 0.80\nbest_f1_train = -1.0\nfor th in CONTAIN_GRID:\n    y_pred = (train_max_ratios >= float(th)).astype(int)\n    f1 = f1_score_binary(train_labels[\"right_place\"].values, y_pred)\n    if f1 > best_f1_train:\n        best_f1_train = float(f1)\n        best_contain_th = float(th)\n\nBEST_CONTAIN_TH = float(best_contain_th)\nprint(\"BEST_CONTAIN_TH (from TRAIN GT):\", BEST_CONTAIN_TH, \"| TRAIN F1:\", best_f1_train)\n\n# ------------------------------------------------------------\n# Helper: compute COCO mAP50:95 on VALID using combined predictions\n# ------------------------------------------------------------\ndef coco_map_from_predstrings(pred_df: pd.DataFrame, gt_coco_json: Path) -> float:\n    gt_coco_json = Path(gt_coco_json)\n    if not gt_coco_json.exists():\n        raise FileNotFoundError(f\"Missing COCO GT json: {gt_coco_json}\")\n\n    coco_gt = COCO(str(gt_coco_json))\n    imgs = coco_gt.dataset.get(\"images\", [])\n    file_to_id = {im[\"file_name\"]: im[\"id\"] for im in imgs}\n    file_to_wh = {im[\"file_name\"]: (im.get(\"width\", 0), im.get(\"height\", 0)) for im in imgs}\n\n    results = []\n    for _, r in pred_df.iterrows():\n        fn = r[\"filename\"]\n        ps = (r.get(\"prediction_string\", \"\") or \"\").strip()\n        if (not ps) or (fn not in file_to_id):\n            continue\n\n        cls_ids, confs, boxes = parse_predstring(ps)\n        if not cls_ids:\n            continue\n\n        W, H = file_to_wh.get(fn, (0, 0))\n        img_id = file_to_id[fn]\n\n        for cid, score, (xc, yc, bw, bh) in zip(cls_ids, confs, boxes):\n            # convert normalized -> absolute pixel\n            x = (xc - bw / 2.0) * W\n            y = (yc - bh / 2.0) * H\n            w = bw * W\n            h = bh * H\n            if w <= 0 or h <= 0:\n                continue\n            cat_id = YOLO_TO_COCO.get(int(cid), int(cid) + 1)\n            results.append({\n                \"image_id\": int(img_id),\n                \"category_id\": int(cat_id),\n                \"bbox\": [float(x), float(y), float(w), float(h)],\n                \"score\": float(score),\n            })\n\n    if len(results) == 0:\n        return 0.0\n\n    coco_dt = coco_gt.loadRes(results)\n    coco_eval = COCOeval(coco_gt, coco_dt, iouType=\"bbox\")\n    coco_eval.evaluate()\n    coco_eval.accumulate()\n    coco_eval.summarize()\n    # stats[0] is mAP@[0.50:0.95]\n    return float(coco_eval.stats[0])\n\nVAL_GT_COCO_JSON = COCO_DATASET_DIR / \"valid\" / \"_annotations.coco.json\"\nVAL_MAP = coco_map_from_predstrings(val_combined, VAL_GT_COCO_JSON)\nprint(\"VAL mAP50:95 (from combined preds):\", VAL_MAP)\n\n# ------------------------------------------------------------\n# C) Tune \"certainty\" threshold on VALID using predicted boxes\n#     - Only boxes with conf >= certain_th are used for right_place\n#     - Containment uses BEST_CONTAIN_TH found from TRAIN GT\n#     - Score = (mAP + F1)/2\n# ------------------------------------------------------------\nval_eval = val_combined.merge(val_labels, on=\"filename\", how=\"left\")\nval_eval[\"right_place\"] = val_eval[\"right_place\"].fillna(0).astype(int)\n\nCERTAIN_GRID = np.linspace(0.2, 1.0, 101)\n\nbest_certain_th = 0.15\nbest_f1_val = -1.0\nbest_score = -1.0\n\n# for th in CERTAIN_GRID:\n#     y_pred = [\n#         pred_right_place_from_predstring_confident(\n#             (ps or \"\"),\n#             face_cls=FACE_YOLO,\n#             shem_cls=SHEM_YOLO,\n#             contain_th=float(BEST_CONTAIN_TH),\n#             certain_th=float(th),\n#         )\n#         for ps in val_eval[\"prediction_string\"].fillna(\"\").astype(str).tolist()\n#     ]\n#     f1 = f1_score_binary(val_eval[\"right_place\"].values, y_pred)\n#     score = (float(VAL_MAP) + float(f1)) / 2.0\n\n#     if score > best_score:\n#         best_score = float(score)\n#         best_f1_val = float(f1)\n#         best_certain_th = float(th)\n\nBEST_CERTAIN_TH = float(best_certain_th)\nprint(\"BEST_CERTAIN_TH (from VAL preds):\", BEST_CERTAIN_TH, \"| VAL F1:\", best_f1_val, \"| VAL SCORE:\", best_score)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Phase 7 — Build submission + visualize test images from both classes","metadata":{}},{"cell_type":"code","source":"# Build submission.csv (filename, prediction_string, right_place)\nsub = test_combined.copy()\n\n# Kaggle format: use \"-\" when there are no boxes\nps = sub[\"prediction_string\"].fillna(\"\").astype(str).str.strip()\nsub[\"prediction_string\"] = ps.mask(ps.eq(\"\"), \"-\")\n\n# right_place should be computed as if empty => no detections\ndef _ps_for_right_place(s: str) -> str:\n    s = (s or \"\").strip()\n    return \"\" if s == \"-\" else s\n\nsub[\"right_place\"] = [\n    pred_right_place_from_predstring_confident(\n        _ps_for_right_place(s),\n        face_cls=FACE_YOLO,\n        shem_cls=SHEM_YOLO,\n        contain_th=float(BEST_CONTAIN_TH),\n        certain_th=float(BEST_CERTAIN_TH),\n    )\n    for s in sub[\"prediction_string\"]\n]\n\nsub.to_csv(\"submission.csv\", index=False)\nprint(\"Wrote: submission.csv\")\ndisplay(sub.head())\n\nprint(\"right_place distribution on TEST:\")\ndisplay(sub[\"right_place\"].value_counts())","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Visualize a few test images from BOTH predicted classes (right_place = 0 and = 1)\nN_SHOW = 800\n\npos = sub[sub[\"right_place\"] == 1].head(N_SHOW)\nneg = sub[sub[\"right_place\"] == 0].head(N_SHOW)\n\nprint(\"Showing right_place=1 examples:\")\nfor _, r in pos.iterrows():\n    plot_image_with_boxes(\n        TEST_IMAGES_DIR / r[\"filename\"],\n        r[\"prediction_string\"],\n        title=f\"TEST {r['filename']} | right_place=1 | contain_th={BEST_CONTAIN_TH:.2f}\"\n    )\n\nprint(\"Showing right_place=0 examples:\")\nfor _, r in neg.iterrows():\n    plot_image_with_boxes(\n        TEST_IMAGES_DIR / r[\"filename\"],\n        r[\"prediction_string\"],\n        title=f\"TEST {r['filename']} | right_place=0 | contain_th={BEST_CONTAIN_TH:.2f}\"\n    )","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}