# %%
import os
import cv2
import torch
import [Link] as nn
import [Link] as F
from [Link] import Dataset, DataLoader
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from [Link] import tqdm
# %%
def sample_frames(video_path, num_frames=16, size=128):
cap = [Link](video_path)
total_frames = int([Link](cv2.CAP_PROP_FRAME_COUNT))
idxs = [Link](0, total_frames-1, num_frames, dtype=int)
frames = []
for i in range(total_frames):
ret, frame = [Link]()
if not ret: break
if i in idxs:
frame = [Link](frame, (size, size))
frame = frame[..., ::-1] / 255.0 # BGR→RGB, normalize
[Link](frame)
[Link]()
frames = [Link](frames) # (T,H,W,C)
assert len(frames) >= num_frames
return frames
# %%
class VideoDataset(Dataset):
def __init__(self, video_dir, df, num_frames=16, size=128, is_test=False):
self.video_dir = video_dir
[Link] = df
self.num_frames = num_frames
[Link] = size
self.is_test = is_test
def __len__(self):
return len([Link])
def __getitem__(self, idx):
row = [Link][idx]
video_path = [Link](self.video_dir, row['filename'])
X = sample_frames(video_path, self.num_frames, [Link])
X = [Link](X, dtype=torch.float32).permute(0,3,1,2) # (T,C,H,W)
if self.is_test:
return X, row['filename']
y = [Link](row['label'], dtype=[Link])
return X, y
# %%
import torch
import [Link] as nn
import [Link] as F
class Residual3DBlock([Link]):
def __init__(self, in_channels, out_channels, stride=1):
super().__init__()
self.conv1 = nn.Conv3d(in_channels, out_channels, kernel_size=3,
stride=stride, padding=1, bias=False)
self.bn1 = nn.BatchNorm3d(out_channels)
self.conv2 = nn.Conv3d(out_channels, out_channels, kernel_size=3,
padding=1, bias=False)
self.bn2 = nn.BatchNorm3d(out_channels)
# projection if in/out channels mismatch or stride > 1
[Link] = [Link]()
if stride != 1 or in_channels != out_channels:
[Link] = [Link](
nn.Conv3d(in_channels, out_channels, kernel_size=1, stride=stride,
bias=False),
nn.BatchNorm3d(out_channels)
)
def forward(self, x):
out = [Link](self.bn1(self.conv1(x)))
out = self.bn2(self.conv2(out))
out += [Link](x)
return [Link](out)
class Residual3DCNN([Link]):
def __init__(self, num_classes=20):
super().__init__()
self.layer1 = Residual3DBlock(3, 32)
self.pool1 = nn.MaxPool3d((1,2,2)) # downsample H,W
self.layer2 = Residual3DBlock(32, 64)
self.pool2 = nn.MaxPool3d((1,2,2))
self.global_pool = nn.AdaptiveAvgPool3d((None, 4, 4)) # keep T, shrink H,W
to 4x4
self.fc1 = [Link](64 * 16 * 4 * 4, 256) # assumes T=16 fixed
self.fc2 = [Link](256, num_classes)
def forward(self, x): # x: (B,T,C,H,W)
x = [Link](0, 2, 1, 3, 4) # -> (B,C,T,H,W)
x = self.layer1(x)
x = self.pool1(x)
x = self.layer2(x)
x = self.pool2(x)
x = self.global_pool(x) # (B,64,T,4,4)
x = [Link]([Link](0), -1) # flatten
x = [Link](self.fc1(x))
return self.fc2(x)
# %%
df = pd.read_csv("private_train/private_train_label.csv")
train_df, val_df = train_test_split(df, test_size=0.2, stratify=df['label'],
random_state=42)
train_dataset = VideoDataset("private_train", train_df)
val_dataset = VideoDataset("private_train", val_df)
train_loader = DataLoader(train_dataset, batch_size=4, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=4, shuffle=False)
# %%
# df = pd.read_csv("private_train/private_train_label.csv")
# _, df = train_test_split(df, test_size=0.1, stratify=df['label'],
random_state=42)
# train_dataset = VideoDataset("private_train", df)
# train_loader = DataLoader(train_dataset, batch_size=4, shuffle=True)
# %%
len(df), len(train_dataset), len(val_dataset), len(train_loader), len(val_loader)
# %%
# len(df), len(train_dataset), len(train_loader)
# %%
# from [Link] import f1_score
# device = [Link]("cuda" if [Link].is_available() else "cpu")
# model = Residual3DCNN().to(device)
# criterion = [Link]()
# optimizer = [Link]([Link](), lr=1e-3)
# best_acc = 0
# epochs = 50
# for epoch in range(1, epochs + 1):
# # Train
# [Link]()
# total_loss = 0
# correct = 0
# for X, y in tqdm(train_loader):
# X, y = [Link](device), [Link](device)
# optimizer.zero_grad()
# preds = model(X)
# loss = criterion(preds, y)
# [Link]()
# [Link]()
# total_loss += [Link]()
# correct += ([Link](1) == y).sum().item()
# acc = correct / len(train_dataset)
# print(f"Epoch {epoch} | train loss {total_loss/len(train_loader):.4f} | acc
{acc:.3f}")
# %%
from [Link] import f1_score
device = [Link]("cuda" if [Link].is_available() else "cpu")
model = Residual3DCNN().to(device)
criterion = [Link]()
optimizer = [Link]([Link](), lr=1e-3)
best_acc = 0
epochs = 50
MODEL_PATH = 'cnn_model.pt'
for epoch in range(1, epochs + 1):
# Train
[Link]()
total_loss = 0
correct = 0
for X, y in tqdm(train_loader):
X, y = [Link](device), [Link](device)
optimizer.zero_grad()
preds = model(X)
loss = criterion(preds, y)
[Link]()
[Link]()
total_loss += [Link]()
correct += ([Link](1) == y).sum().item()
acc = correct / len(train_dataset)
print(f"Epoch {epoch} | train loss {total_loss/len(train_loader):.4f} | acc
{acc:.3f}")
# Val
[Link]()
val_loss, val_correct = 0, 0
all_preds = []
all_targets = []
with torch.no_grad():
for X, y in val_loader:
X, y = [Link](device), [Link](device)
preds = model(X)
val_loss += criterion(preds, y).item()
val_correct += ([Link](1) == y).sum().item()
all_preds.extend([Link](1).cpu().numpy())
all_targets.extend([Link]().numpy())
val_acc = val_correct / len(val_dataset)
f1 = f1_score(all_targets, all_preds, average='weighted') # or 'macro',
'micro', etc.
print(f"Val loss {val_loss/len(val_loader):.4f} | acc {val_acc:.3f} | F1
{f1:.4f}")
if best_acc < val_acc:
best_acc = val_acc
[Link](model.state_dict(), MODEL_PATH)
print('SAVED MODEL')
# %%
model.load_state_dict([Link](MODEL_PATH))
test_files = sorted([Link]("private_test"), key=lambda x: int([Link]('.')[0]))
test_df = [Link]({"filename": test_files})
test_dataset = VideoDataset("private_test", test_df, is_test=True)
test_loader = DataLoader(test_dataset, batch_size=1, shuffle=False)
[Link]()
preds = []
with torch.no_grad():
for X, fname in tqdm(test_loader):
X = [Link](device)
out = model(X)
pred = [Link](1).item()
[Link]((fname[0], pred))
[Link](preds, columns=["filename","label"]).to_csv("[Link]",
index=False)