{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":3004,"databundleVersionId":861823,"sourceType":"competition"}],"dockerImageVersionId":30747,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import torch\nimport numpy as np\nfrom PIL import Image\nimport torch.nn as nn\nimport torch.optim as optim\nfrom torch.utils.data import DataLoader, Dataset\nfrom sklearn.model_selection import train_test_split\nfrom torchvision import transforms\nfrom collections import deque\nfrom tqdm import tqdm\nimport polars as pl\nimport timm","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Custom dataset class\nclass ImageCSVLoader(Dataset):\n    def __init__(self, data_frame, transform=None):\n        self.data_frame = data_frame\n        self.transform = transform\n\n    def __len__(self):\n        return len(self.data_frame)\n\n    def __getitem__(self, idx):\n        label = int(self.data_frame[idx, 0])\n        pixels = self.data_frame[idx, 1:].astype(np.uint8)\n        image = Image.fromarray(pixels.reshape(28, 28))\n        return self.transform(image), label\n\nclass ImageCSVLoaderTest(Dataset):\n    def __init__(self, data_frame, transform=None):\n        self.data_frame = data_frame\n        self.transform = transform\n\n    def __len__(self):\n        return len(self.data_frame)\n\n    def __getitem__(self, idx):\n        pixels = self.data_frame[idx, :].astype(np.uint8)\n        image = Image.fromarray(pixels.reshape(28, 28))\n        return self.transform(image)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Training function\ndef train(model, loader, criterion, optimizer, scheduler):\n    model.train()\n    running_loss = 0.0\n    losses = deque(maxlen=24)\n    progress_bar = tqdm(loader, desc='Training', leave=True)\n    \n    for images, labels in progress_bar:\n        images, labels = images.cuda(), labels.cuda()\n\n        optimizer.zero_grad()\n        outputs = model(images)\n        loss = criterion(outputs, labels)\n        loss.backward()\n        optimizer.step()\n\n        running_loss += loss.item()\n        losses.append(loss.item())\n        \n        avg_loss = sum(losses) / len(losses)\n        progress_bar.set_postfix(loss=avg_loss, lr=optimizer.param_groups[0]['lr'])\n    \n    scheduler.step()\n    return running_loss / len(loader)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Validation function\ndef validate(model, loader, criterion):\n    model.eval()\n    running_loss = 0.0\n    correct = 0\n    total = 0\n    losses = deque(maxlen=64)\n    progress_bar = tqdm(loader, desc='Validating', leave=False)\n    \n    with torch.no_grad():\n        for images, labels in progress_bar:\n            images, labels = images.cuda(), labels.cuda()\n\n            outputs = model(images)\n            loss = criterion(outputs, labels)\n            running_loss += loss.item()\n            losses.append(loss.item())\n\n            _, predicted = torch.max(outputs, 1)\n            total += labels.size(0)\n            correct += (predicted == labels).sum().item()\n\n            avg_loss = sum(losses) / len(losses)\n            progress_bar.set_postfix(loss=avg_loss)\n    \n    accuracy = correct / total\n    return running_loss / len(loader), accuracy","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Inference function for test data\ndef inference(model, loader):\n    model.eval()\n    predictions = []\n    progress_bar = tqdm(loader, desc='Inferencing', leave=False)\n    \n    with torch.no_grad():\n        for images in progress_bar:\n            images = images.cuda()\n            outputs = model(images)\n            _, predicted = torch.max(outputs, 1)\n            predictions.extend(predicted.cpu().numpy())\n    \n    return predictions","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Main script entry point\nif __name__ == '__main__':\n    # Parameters\n    input_folder = '/kaggle/input/digit-recognizer'\n    train_csv_file = f'{input_folder}/train.csv'\n    test_csv_file = f'{input_folder}/test.csv'\n\n    num_workers = 4\n    input_size = 28\n    batch_size = 8\n    num_epochs = 20\n    validation_split = 0.025\n\n    dropout = 0.5\n    learning_rate = 1e-4\n    weight_decay = 1e-5\n    gamma = 0.925\n\n    # Transformations\n    train_transform = transforms.Compose([\n        transforms.Grayscale(num_output_channels=3),\n        transforms.RandomAffine(degrees=10, translate=(0.1, 0.1)),\n        transforms.ToTensor(),\n        transforms.Normalize((0.1307,), (0.3081,))\n    ])\n\n    test_transform = transforms.Compose([\n        transforms.Grayscale(num_output_channels=3),\n        transforms.ToTensor(),\n        transforms.Normalize((0.1307,), (0.3081,))\n    ])\n\n    # Load the CSV data\n    data_frame = pl.read_csv(train_csv_file)\n\n    # Split the data into train and validation sets\n    train_data, val_data = train_test_split(data_frame.to_numpy(), test_size=validation_split, random_state=42)\n\n    # Create datasets\n    train_dataset = ImageCSVLoader(data_frame=train_data, transform=train_transform)\n    val_dataset = ImageCSVLoader(data_frame=val_data, transform=test_transform)\n\n    # Load test data\n    test_data_frame = pl.read_csv(test_csv_file).to_numpy()\n\n    # Create test dataset\n    test_dataset = ImageCSVLoaderTest(data_frame=test_data_frame, transform=test_transform)\n\n    # Data loaders\n    train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, num_workers=num_workers, pin_memory=True)\n    val_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False, num_workers=num_workers, pin_memory=True)\n    test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False, num_workers=num_workers, pin_memory=True)\n\n    # Model\n    model = timm.create_model('resnet34d', pretrained=False, num_classes=10, drop_rate=dropout)\n    model = model.cuda()\n\n    # Print the number of parameters\n    print(f'Total number of trainable parameters: {sum(p.numel() for p in model.parameters() if p.requires_grad)}')\n\n    # Loss and optimizer\n    criterion = nn.CrossEntropyLoss()\n    optimizer = optim.AdamW(model.parameters(), lr=learning_rate, weight_decay=weight_decay)\n    scheduler = optim.lr_scheduler.ExponentialLR(optimizer, gamma=gamma)\n\n    # Initialize best model tracking\n    best_val_loss = float('inf')\n    best_model_wts = model.state_dict()\n\n    for epoch in range(num_epochs):\n        print(f'Epoch {epoch + 1}/{num_epochs}')\n        train_loss = train(model, train_loader, criterion, optimizer, scheduler)\n        val_loss, val_accuracy = validate(model, val_loader, criterion)\n        print(f'Validation Loss: {val_loss:.4f}, Validation Accuracy: {val_accuracy:.4f}')\n        \n        # Save best model based on validation loss\n        if val_loss < best_val_loss:\n            best_val_loss = val_loss\n            best_model_wts = model.state_dict()\n\n    # Load best model weights\n    model.load_state_dict(best_model_wts)\n    print('Training complete')\n    \n    # Inference on test data\n    test_predictions = inference(model, test_loader)\n\n    # Save predictions to CSV\n    image_ids = np.arange(1, len(test_predictions) + 1)\n    test_df = pl.DataFrame({\n        'ImageId': image_ids,\n        'Label': test_predictions\n    })\n\n    test_df.write_csv('/kaggle/working/submission.csv')\n    print('Inference complete and results saved.')","metadata":{},"execution_count":null,"outputs":[]}]}