{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.12","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[{"sourceId":91498,"databundleVersionId":11655853,"sourceType":"competition"}],"dockerImageVersionId":30918,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"# Import necessary libraries\nimport os\nimport numpy as np\nimport pandas as pd\nimport torch\nimport torchvision.transforms as transforms\nfrom PIL import Image\nimport cv2\nfrom sklearn.decomposition import PCA\nfrom sklearn.cluster import KMeans\nfrom tqdm import tqdm\nfrom collections import defaultdict\n\n# Set up paths\nTEST_DIR = '/kaggle/input/image-matching-challenge-2025/test'\nSUBMISSION_PATH = '/kaggle/working/submission.csv'\n\n# Device configuration\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n\n# Scan all test images and organize by dataset\ndef scan_test_images():\n    images_by_dataset = defaultdict(list)\n    for root, _, files in os.walk(TEST_DIR):\n        for filename in files:\n            if filename.lower().endswith(('.jpg', '.jpeg', '.png')):\n                full_path = os.path.join(root, filename)\n                rel_path = os.path.relpath(full_path, TEST_DIR)\n                parts = rel_path.split(os.sep)\n                if parts:\n                    dataset_name = parts[0]\n                    images_by_dataset[dataset_name].append({\n                        'full_path': full_path,\n                        'rel_path': rel_path,\n                        'filename': filename\n                    })\n    return images_by_dataset\n\n# SIFT feature extractor\ndef extract_sift_features(image_path):\n    img = cv2.imread(image_path, cv2.IMREAD_GRAYSCALE)\n    if img is None:\n        return None\n    sift = cv2.SIFT_create()\n    _, descriptors = sift.detectAndCompute(img, None)\n    if descriptors is None:\n        return None\n    if descriptors.shape[0] > 100:\n        descriptors = descriptors[np.random.choice(descriptors.shape[0], 100, replace=False)]\n    elif descriptors.shape[0] < 100:\n        padding = np.zeros((100 - descriptors.shape[0], descriptors.shape[1]))\n        descriptors = np.vstack([descriptors, padding])\n    return descriptors.flatten()\n\n# Feature extraction pipeline\ndef extract_features(image_paths):\n    features = []\n    valid_indices = []\n    for i, path in enumerate(tqdm(image_paths)):\n        feat = extract_sift_features(path)\n        if feat is not None:\n            features.append(feat)\n            valid_indices.append(i)\n    return np.array(features), valid_indices\n\n# Pose estimation using essential matrix\ndef estimate_relative_pose(img1_path, img2_path, focal=1000, pp=(512, 512)):\n    img1 = cv2.imread(img1_path, cv2.IMREAD_GRAYSCALE)\n    img2 = cv2.imread(img2_path, cv2.IMREAD_GRAYSCALE)\n    if img1 is None or img2 is None:\n        return None, None\n    sift = cv2.SIFT_create()\n    kp1, des1 = sift.detectAndCompute(img1, None)\n    kp2, des2 = sift.detectAndCompute(img2, None)\n    if des1 is None or des2 is None or len(des1) < 8 or len(des2) < 8:\n        return None, None\n    bf = cv2.BFMatcher()\n    matches = bf.knnMatch(des1, des2, k=2)\n    good = [m for m, n in matches if m.distance < 0.75 * n.distance]\n    if len(good) < 8:\n        return None, None\n    pts1 = np.float32([kp1[m.queryIdx].pt for m in good])\n    pts2 = np.float32([kp2[m.trainIdx].pt for m in good])\n    E, _ = cv2.findEssentialMat(pts1, pts2, focal=focal, pp=pp, method=cv2.RANSAC)\n    if E is None:\n        return None, None\n    _, R, t, _ = cv2.recoverPose(E, pts1, pts2, focal=focal, pp=pp)\n    return R, t\n\n# Structure from Motion on cluster\n\ndef perform_sfm(images):\n    if len(images) <= 1:\n        return [None]*len(images), [None]*len(images)\n    R_list = [np.eye(3)]\n    t_list = [np.zeros(3)]\n    ref_path = images[0]['full_path']\n    for i in range(1, len(images)):\n        R, t = estimate_relative_pose(ref_path, images[i]['full_path'])\n        R_list.append(R if R is not None else None)\n        t_list.append(t.flatten() if t is not None else None)\n    return R_list, t_list\n\n# Format R and t for submission\ndef format_pose(R, t):\n    try:\n        R_str = ';'.join([f\"{x:.6f}\" for x in R.flatten()])\n        t_str = ';'.join([f\"{x:.6f}\" for x in t])\n    except:\n        R_str = 'nan;nan;nan;nan;nan;nan;nan;nan;nan'\n        t_str = 'nan;nan;nan'\n    return R_str, t_str\n\n# Validate R and t string lengths and content\ndef validate_pose(pose_str, length):\n    try:\n        parts = [float(x) for x in pose_str.split(';')]\n        return len(parts) == length\n    except:\n        return False\n\n# Main execution function\ndef main():\n    print(\"Starting pipeline...\")\n    images_by_dataset = scan_test_images()\n    all_rows = []\n    for dataset, images in images_by_dataset.items():\n        print(f\"Processing dataset: {dataset} with {len(images)} images\")\n        image_paths = [img['full_path'] for img in images]\n        features, valid_idx = extract_features(image_paths)\n        if features.shape[0] == 0:\n            continue\n        n_components = min(16, features.shape[0], features.shape[1])\n        pca = PCA(n_components=n_components)\n        reduced = pca.fit_transform(features)\n        k = max(2, min(10, len(reduced)//20))\n        kmeans = KMeans(n_clusters=k, random_state=42).fit(reduced)\n        labels = kmeans.labels_\n        for cid in range(k):\n            indices = [i for i, label in zip(valid_idx, labels) if label == cid]\n            if not indices:\n                continue\n            cluster_images = [images[i] for i in indices]\n            R_list, t_list = perform_sfm(cluster_images)\n            for img, R, t in zip(cluster_images, R_list, t_list):\n                R_str, t_str = format_pose(R, t)\n                all_rows.append({\n                    'dataset': dataset,\n                    'scene': f\"cluster{cid}\",\n                    'image': img['filename'],\n                    'rotation_matrix': R_str,\n                    'translation_vector': t_str\n                })\n\n    # Final DataFrame\n    df = pd.DataFrame(all_rows)\n    df = df[['dataset', 'scene', 'image', 'rotation_matrix', 'translation_vector']]\n    df = df[df['rotation_matrix'].apply(lambda x: validate_pose(x, 9))]\n    df = df[df['translation_vector'].apply(lambda x: validate_pose(x, 3))]\n    df.to_csv(SUBMISSION_PATH, index=False)\n    print(f\"Submission saved: {SUBMISSION_PATH} with {len(df)} rows\")\n\nif __name__ == \"__main__\":\n    main()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-08T22:45:03.136546Z","iopub.execute_input":"2025-04-08T22:45:03.136865Z","iopub.status.idle":"2025-04-08T22:46:09.184089Z","shell.execute_reply.started":"2025-04-08T22:45:03.136831Z","shell.execute_reply":"2025-04-08T22:46:09.182918Z"}},"outputs":[],"execution_count":null}]}