{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.11.11","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":92399,"databundleVersionId":11038207,"sourceType":"competition"},{"sourceId":375539,"sourceType":"modelInstanceVersion","isSourceIdPinned":true,"modelInstanceId":310349,"modelId":330730}],"dockerImageVersionId":31011,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# Dashcam Collision Prediction: mvit_v2_s inference based on best model\n\n## Stupidly enough, I trained on jpegs, so code saves and loads a jpeg for the extracted images from video, so that it gets exact same results...\n","metadata":{}},{"cell_type":"code","source":"#!/usr/bin/env python3\n# -*- coding: utf-8 -*-\n\nimport os\nimport cv2\nimport numpy as np\nimport math\nimport pandas as pd\nimport torch\nimport torch.nn as nn\nfrom torchvision.models.video import mvit_v2_s, MViT_V2_S_Weights\nfrom tqdm.notebook import tqdm\nfrom pathlib import Path\nfrom torchvision import transforms\nfrom PIL import Image\n\n# --- CONFIG ---\nCHECKPOINT_PATH = '/kaggle/input/bestmodelnexarchallenge/pytorch/default/1/del025weight_61_acc_93.65_CP_81.80_CR_52.01_NCP_94.51_NCR_98.62.pt' \nTEST_CSV = '/kaggle/input/nexar-collision-prediction/test.csv'\nTEST_DIR = '/kaggle/input/nexar-collision-prediction/test/'\nDEVICE = torch.device('cuda' if torch.cuda.is_available() else 'cpu')\nFRAME_SIZE = (256, 256) #Initial scaling from video\nNUM_FRAMES = 16\nFRAME_STEP = 4\n\nclass NewCrashPredictor(nn.Module):\n    def __init__(self, num_frames=16):\n        super().__init__()\n\n        self.num_frames = num_frames\n\n        # Load pretrained backbone with proper weights\n        self.backbone_rgb = mvit_v2_s(weights=MViT_V2_S_Weights.DEFAULT)\n        self.backbone_rgb.head = nn.Identity()\n\n        # Output dimension from MViT V2 Small\n        hidden_size = 768\n\n        # Binary classifier for event occurrence\n        self.event_classifier = nn.Sequential(\n            nn.Linear(hidden_size, 256),\n            nn.ReLU(),\n            nn.Dropout(0.3),\n            nn.Linear(256, 64),\n            nn.ReLU(),\n            nn.Dropout(0.2),\n            nn.Linear(64, 1)  # Binary output: will event occur or not\n        )\n\n    def forward(self, x, return_embeddings=False):\n        # Process through modified backbone\n        features = self.backbone_rgb(x)\n\n        # Stage 1: Predict if event will occur (return logits, not probability)\n        event_logits = self.event_classifier(features) \n\n        if return_embeddings:\n            return event_logits,features\n\n        return event_logits\n\n# --- FRAME EXTRACTION (as in preprocess_videos.py) ---\ndef extract_last_frames(path, num_frames=16, step=4):\n    cap = cv2.VideoCapture(path)\n    total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))\n    if total < (num_frames-1)*step+1:\n        indices = [0]*num_frames\n    else:\n        indices = [total - 1 - i*step for i in reversed(range(num_frames))]\n\n    preprocess = transforms.Compose([\n        transforms.Resize((224, 224)), \n        transforms.ToTensor(),\n        transforms.Normalize(mean=[0.45, 0.45, 0.45], std=[0.225, 0.225, 0.225])\n    ])\n   \n    frames = []\n    for idx in tqdm(indices, desc=f\"Extracting frames from {os.path.basename(path)}\", leave=False):\n        cap.set(cv2.CAP_PROP_POS_FRAMES, idx)\n        ret, frame = cap.read()\n        if not ret:\n            frame = np.zeros((256, 256, 3), dtype=np.uint8)\n        else:\n            frame = cv2.resize(frame, (256, 256), interpolation=cv2.INTER_LANCZOS4)\n\n        # Stupid - Save frame to temp file and load again as jpeg to get exact same results as other/submitted inference code.\n        frame_path = f\"temp.jpg\"\n        cv2.imwrite(frame_path, frame)\n        #load image again\n        frame = Image.open(frame_path)\n        frame = preprocess(frame)\n        frames.append(frame)\n\n    cap.release()\n    frames = torch.stack(frames, dim=1)  # (C, T, H, W)\n    return frames.unsqueeze(0)  # (1, C, T, H, W)\n\n# --- LOAD MODEL\nprint (\"Loading model...\")\nmodel = NewCrashPredictor()\ncheckpoint = torch.load(CHECKPOINT_PATH, weights_only=True)\nmodel.load_state_dict(checkpoint['model_state_dict'], strict=False)\nmodel = model.to(DEVICE)\nmodel.eval()\n\n# --- INFERENCE ---\ndf_test = pd.read_csv(TEST_CSV)\ndf_test['id'] = df_test['id'].astype(str).str.zfill(5)\nresults = []\n\nwith torch.no_grad():\n    for vid in tqdm(df_test['id'], desc=\"Inference\"):\n        video_path = os.path.join(TEST_DIR, f\"{vid}.mp4\")\n        frames = extract_last_frames(video_path) #load_last_frames_from_folder(f\"test_frames256/{vid}\")\n        frames = frames.to(DEVICE)\n        logits = model(frames)\n        prob = torch.sigmoid(logits).cpu().numpy().flatten()[0]\n        print (f\"Video {vid} has probability {prob} of event occurring\")\n        results.append({'id': vid, 'score': prob})\n\nsubmission = pd.DataFrame(results)\nsubmission.to_csv('submission.csv', index=False)\nprint(\"✅ Written → submission.csv\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-05-06T02:30:10.904890Z","iopub.execute_input":"2025-05-06T02:30:10.905504Z","iopub.status.idle":"2025-05-06T04:06:52.138443Z","shell.execute_reply.started":"2025-05-06T02:30:10.905479Z","shell.execute_reply":"2025-05-06T04:06:52.137841Z"}},"outputs":[],"execution_count":null}]}