{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":3004,"databundleVersionId":861823,"sourceType":"competition"}],"dockerImageVersionId":30733,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"# This Python 3 environment comes with many helpful analytics libraries installed\n# It is defined by the kaggle/python Docker image: https://github.com/kaggle/docker-python\n# For example, here's several helpful packages to load\n\nimport numpy as np # linear algebra\nimport pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\n\n# Input data files are available in the read-only \"../input/\" directory\n# For example, running this (by clicking run or pressing Shift+Enter) will list all files under the input directory\n\nimport os\nfor dirname, _, filenames in os.walk('/kaggle/input'):\n    for filename in filenames:\n        print(os.path.join(dirname, filename))\n\n# You can write up to 20GB to the current directory (/kaggle/working/) that gets preserved as output when you create a version using \"Save & Run All\" \n# You can also write temporary files to /kaggle/temp/, but they won't be saved outside of the current session","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","execution":{"iopub.status.busy":"2024-06-21T19:55:13.032517Z","iopub.execute_input":"2024-06-21T19:55:13.033392Z","iopub.status.idle":"2024-06-21T19:55:13.409138Z","shell.execute_reply.started":"2024-06-21T19:55:13.033351Z","shell.execute_reply":"2024-06-21T19:55:13.408294Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"!pip install torch-summary","metadata":{"execution":{"iopub.status.busy":"2024-06-21T19:55:13.41076Z","iopub.execute_input":"2024-06-21T19:55:13.411115Z","iopub.status.idle":"2024-06-21T19:55:26.643887Z","shell.execute_reply.started":"2024-06-21T19:55:13.41109Z","shell.execute_reply":"2024-06-21T19:55:26.642855Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nimport torch.optim as optim\nfrom torchvision import datasets, transforms\nfrom torch.utils.data import DataLoader\nimport math\nimport matplotlib.pyplot as plt\nfrom torchsummary import summary\nimport pandas as pd\nimport numpy as np\nfrom torch.utils.data import Dataset","metadata":{"execution":{"iopub.status.busy":"2024-06-21T19:55:26.645331Z","iopub.execute_input":"2024-06-21T19:55:26.645642Z","iopub.status.idle":"2024-06-21T19:55:31.394923Z","shell.execute_reply.started":"2024-06-21T19:55:26.645615Z","shell.execute_reply":"2024-06-21T19:55:31.394123Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"class KANLinear(torch.nn.Module):\n    def __init__(self, in_features, out_features, grid_size=5, spline_order=5, scale_noise=0.15, scale_base=1.0,\n                 scale_spline=1.0, enable_standalone_scale_spline=True, base_activation=torch.nn.SiLU, grid_eps=0.02,\n                 grid_range=[-1, 1]):\n        super(KANLinear, self).__init__()\n        self.in_features = in_features\n        self.out_features = out_features\n        self.grid_size = grid_size\n        self.spline_order = spline_order\n\n        h = (grid_range[1] - grid_range[0]) / grid_size\n        grid = (\n            (\n                torch.arange(-spline_order, grid_size + spline_order + 1) * h\n                + grid_range[0]\n            )\n            .expand(in_features, -1)\n            .contiguous()\n        )\n        self.register_buffer(\"grid\", grid)\n\n        self.base_weight = torch.nn.Parameter(torch.Tensor(out_features, in_features))\n        self.spline_weight = torch.nn.Parameter(\n            torch.Tensor(out_features, in_features, grid_size + spline_order)\n        )\n        if enable_standalone_scale_spline:\n            self.spline_scaler = torch.nn.Parameter(\n                torch.Tensor(out_features, in_features)\n            )\n\n        self.scale_noise = scale_noise\n        self.scale_base = scale_base\n        self.scale_spline = scale_spline\n        self.enable_standalone_scale_spline = enable_standalone_scale_spline\n        self.base_activation = base_activation()\n        self.grid_eps = grid_eps\n\n        self.reset_parameters()\n\n    def reset_parameters(self):\n        torch.nn.init.kaiming_uniform_(self.base_weight, a=math.sqrt(5) * self.scale_base)\n        with torch.no_grad():\n            noise = (\n                (\n                    torch.rand(self.grid_size + 1, self.in_features, self.out_features)\n                    - 1 / 2\n                )\n                * self.scale_noise\n                / self.grid_size\n            )\n            self.spline_weight.data.copy_(\n                (self.scale_spline if not self.enable_standalone_scale_spline else 1.0)\n                * self.curve2coeff(\n                    self.grid.T[self.spline_order : -self.spline_order],\n                    noise,\n                )\n            )\n            if self.enable_standalone_scale_spline:\n                # torch.nn.init.constant_(self.spline_scaler, self.scale_spline)\n                torch.nn.init.kaiming_uniform_(self.spline_scaler, a=math.sqrt(5) * self.scale_spline)\n\n    def b_splines(self, x: torch.Tensor):\n        \"\"\"\n        Compute the B-spline bases for the given input tensor.\n\n        Args:\n            x (torch.Tensor): Input tensor of shape (batch_size, in_features).\n\n        Returns:\n            torch.Tensor: B-spline bases tensor of shape (batch_size, in_features, grid_size + spline_order).\n        \"\"\"\n        assert x.dim() == 2 and x.size(1) == self.in_features\n\n        grid: torch.Tensor = (\n            self.grid\n        )  # (in_features, grid_size + 2 * spline_order + 1)\n        x = x.unsqueeze(-1)\n        bases = ((x >= grid[:, :-1]) & (x < grid[:, 1:])).to(x.dtype)\n        for k in range(1, self.spline_order + 1):\n            bases = (\n                (x - grid[:, : -(k + 1)])\n                / (grid[:, k:-1] - grid[:, : -(k + 1)])\n                * bases[:, :, :-1]\n            ) + (\n                (grid[:, k + 1 :] - x)\n                / (grid[:, k + 1 :] - grid[:, 1:(-k)])\n                * bases[:, :, 1:]\n            )\n\n        assert bases.size() == (\n            x.size(0),\n            self.in_features,\n            self.grid_size + self.spline_order,\n        )\n        return bases.contiguous()\n\n    def curve2coeff(self, x: torch.Tensor, y: torch.Tensor):\n        \"\"\"\n        Compute the coefficients of the curve that interpolates the given points.\n\n        Args:\n            x (torch.Tensor): Input tensor of shape (batch_size, in_features).\n            y (torch.Tensor): Output tensor of shape (batch_size, in_features, out_features).\n\n        Returns:\n            torch.Tensor: Coefficients tensor of shape (out_features, in_features, grid_size + spline_order).\n        \"\"\"\n        assert x.dim() == 2 and x.size(1) == self.in_features\n        assert y.size() == (x.size(0), self.in_features, self.out_features)\n\n        A = self.b_splines(x).transpose(\n            0, 1\n        )  # (in_features, batch_size, grid_size + spline_order)\n        B = y.transpose(0, 1)  # (in_features, batch_size, out_features)\n        solution = torch.linalg.lstsq(\n            A, B\n        ).solution  # (in_features, grid_size + spline_order, out_features)\n        result = solution.permute(\n            2, 0, 1\n        )  # (out_features, in_features, grid_size + spline_order)\n\n        assert result.size() == (\n            self.out_features,\n            self.in_features,\n            self.grid_size + self.spline_order,\n        )\n        return result.contiguous()\n\n    @property\n    def scaled_spline_weight(self):\n        return self.spline_weight * (\n            self.spline_scaler.unsqueeze(-1)\n            if self.enable_standalone_scale_spline\n            else 1.0\n        )\n\n    def forward(self, x: torch.Tensor):\n        assert x.dim() == 2 and x.size(1) == self.in_features\n\n        base_output = F.linear(self.base_activation(x), self.base_weight)\n        spline_output = F.linear(\n            self.b_splines(x).view(x.size(0), -1),\n            self.scaled_spline_weight.view(self.out_features, -1),\n        )\n        return base_output + spline_output\n\n    @torch.no_grad()\n    def update_grid(self, x: torch.Tensor, margin=0.01):\n        assert x.dim() == 2 and x.size(1) == self.in_features\n        batch = x.size(0)\n\n        splines = self.b_splines(x)  # (batch, in, coeff)\n        splines = splines.permute(1, 0, 2)  # (in, batch, coeff)\n        orig_coeff = self.scaled_spline_weight  # (out, in, coeff)\n        orig_coeff = orig_coeff.permute(1, 2, 0)  # (in, coeff, out)\n        unreduced_spline_output = torch.bmm(splines, orig_coeff)  # (in, batch, out)\n        unreduced_spline_output = unreduced_spline_output.permute(\n            1, 0, 2\n        )  # (batch, in, out)\n\n        # sort each channel individually to collect data distribution\n        x_sorted = torch.sort(x, dim=0)[0]\n        grid_adaptive = x_sorted[\n            torch.linspace(\n                0, batch - 1, self.grid_size + 1, dtype=torch.int64, device=x.device\n            )\n        ]\n\n        uniform_step = (x_sorted[-1] - x_sorted[0] + 2 * margin) / self.grid_size\n        grid_uniform = (\n            torch.arange(\n                self.grid_size + 1, dtype=torch.float32, device=x.device\n            ).unsqueeze(1)\n            * uniform_step\n            + x_sorted[0]\n            - margin\n        )\n\n        grid = self.grid_eps * grid_uniform + (1 - self.grid_eps) * grid_adaptive\n        grid = torch.concatenate(\n            [\n                grid[:1]\n                - uniform_step\n                * torch.arange(self.spline_order, 0, -1, device=x.device).unsqueeze(1),\n                grid,\n                grid[-1:]\n                + uniform_step\n                * torch.arange(1, self.spline_order + 1, device=x.device).unsqueeze(1),\n            ],\n            dim=0,\n        )\n\n        self.grid.copy_(grid.T)\n        self.spline_weight.data.copy_(self.curve2coeff(x, unreduced_spline_output))\n\n    def regularization_loss(self, regularize_activation=1.0, regularize_entropy=1.0):\n        \"\"\"\n        Compute the regularization loss.\n\n        This is a dumb simulation of the original L1 regularization as stated in the\n        paper, since the original one requires computing absolutes and entropy from the\n        expanded (batch, in_features, grid_size + spline_order) sized tensor, which is\n        extremely slow and consumes a lot of memory.\n        \"\"\"\n        weight = self.scaled_spline_weight\n        activation = self.base_activation(self.base_weight)\n        activation_regularization = weight.square().mean()\n        entropy_regularization = (\n            -weight.softmax(dim=-1) * weight.log_softmax(dim=-1)\n        ).mean()\n        regularization = (\n            regularize_activation * activation_regularization\n            + regularize_entropy * entropy_regularization\n        )\n        return regularization\n\n\nclass CNNKAN(nn.Module):\n    def __init__(self):\n        super(CNNKAN, self).__init__()\n        self.conv1 = nn.Conv2d(1, 32, kernel_size=3, padding=1)\n        self.conv2 = nn.Conv2d(32, 64, kernel_size=3, padding=1)\n        self.pool = nn.MaxPool2d(2, 2)\n        self.fc1 = KANLinear(64 * 7 * 7, 128)\n        self.fc2 = KANLinear(128, 10)\n\n    def forward(self, x):\n        x = self.pool(F.relu(self.conv1(x)))\n        x = self.pool(F.relu(self.conv2(x)))\n        x = x.view(-1, 64 * 7 * 7)\n        x = F.relu(self.fc1(x))\n        x = self.fc2(x)\n        return x\n\nclass KaggleMNISTDataset(Dataset):\n    def __init__(self, csv_file, transform=None, train=True):\n        self.data = pd.read_csv(csv_file)\n        self.transform = transform\n        self.train = train\n        if train:\n            self.labels = self.data.iloc[:, 0].values\n            self.images = self.data.iloc[:, 1:].values.reshape(-1, 28, 28).astype('float32')\n        else:\n            self.images = self.data.values.reshape(-1, 28, 28).astype('float32')\n\n    def __len__(self):\n        return len(self.data)\n\n    def __getitem__(self, idx):\n        image = self.images[idx]\n        if self.transform:\n            image = self.transform(image)\n        if self.train:\n            label = self.labels[idx]\n            return image, label\n        else:\n            return image\n","metadata":{"execution":{"iopub.status.busy":"2024-06-21T19:55:31.39699Z","iopub.execute_input":"2024-06-21T19:55:31.397391Z","iopub.status.idle":"2024-06-21T19:55:31.43868Z","shell.execute_reply.started":"2024-06-21T19:55:31.397365Z","shell.execute_reply":"2024-06-21T19:55:31.437733Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def train(model, device, train_loader, optimizer, epoch):\n    model.train()\n    running_loss = 0.0\n    for batch_idx, (data, target) in enumerate(train_loader):\n        data, target = data.to(device), target.to(device)\n        optimizer.zero_grad()\n        output = model(data)\n        loss = F.cross_entropy(output, target)\n        loss.backward()\n        optimizer.step()\n        running_loss += loss.item()\n        if batch_idx % 100 == 99:\n            print(f'Epoch: {epoch} [{batch_idx * len(data)}/{len(train_loader.dataset)} '\n                  f'({100. * batch_idx / len(train_loader):.0f}%)]\\tLoss: {running_loss / 100:.6f}')\n            running_loss = 0.0\ndef evaluate(model, device, test_loader):\n    model.eval()\n    test_loss = 0\n    correct = 0\n    with torch.no_grad():\n        for data, target in test_loader:\n            data, target = data.to(device), target.to(device)\n            output = model(data)\n            test_loss += F.cross_entropy(output, target, reduction='sum').item()\n            pred = output.argmax(dim=1, keepdim=True)\n            correct += pred.eq(target.view_as(pred)).sum().item()\n\n    test_loss /= len(test_loader.dataset)\n    accuracy = 100. * correct / len(test_loader.dataset)\n    print(f'\\nTest set: Average loss: {test_loss:.4f}, Accuracy: {correct}/{len(test_loader.dataset)} '\n          f'({accuracy:.0f}%)\\n')","metadata":{"execution":{"iopub.status.busy":"2024-06-21T19:55:31.439673Z","iopub.execute_input":"2024-06-21T19:55:31.43993Z","iopub.status.idle":"2024-06-21T19:55:31.456788Z","shell.execute_reply.started":"2024-06-21T19:55:31.439908Z","shell.execute_reply":"2024-06-21T19:55:31.455977Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def load_model(model_path, model_class):\n    model = model_class().to(device)\n    model.load_state_dict(torch.load(model_path, map_location=device))\n    model.eval()\n    return model\n\ndef predict_and_create_submission(model, test_csv_path, submission_csv_path):\n    # Define transformations\n    transform = transforms.Compose([transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,))])\n    \n    # Load test dataset\n    test_dataset = KaggleMNISTDataset(csv_file=test_csv_path, transform=transform, train=False)\n    test_loader = DataLoader(test_dataset, batch_size=64, shuffle=False)\n    \n    predictions = []\n    with torch.no_grad():\n        for data in test_loader:\n            data = data.to(device)\n            output = model(data)\n            pred = output.argmax(dim=1, keepdim=True)\n            predictions.extend(pred.cpu().numpy().flatten())\n    \n    # Create a DataFrame with the results\n    image_ids = np.arange(1, len(predictions) + 1)  # Assuming image IDs start from 1\n    submission_df = pd.DataFrame({'ImageId': image_ids, 'Label': predictions})\n    submission_df.to_csv(submission_csv_path, index=False)\n\n\n","metadata":{"execution":{"iopub.status.busy":"2024-06-21T19:55:31.457781Z","iopub.execute_input":"2024-06-21T19:55:31.458073Z","iopub.status.idle":"2024-06-21T19:55:31.47042Z","shell.execute_reply.started":"2024-06-21T19:55:31.458043Z","shell.execute_reply":"2024-06-21T19:55:31.469591Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def main():\n    use_cuda = torch.cuda.is_available()\n    global device\n    device = torch.device(\"cuda\" if use_cuda else \"cpu\")\n\n    # Initialize the model\n    model_path = \"mnist_cnn_kan.pth\"  # Path to the saved model\n\n    # Check if model exists; if not, train and save it\n    try:\n        model = load_model(model_path, CNNKAN)\n    except FileNotFoundError:\n        model = CNNKAN().to(device)\n        # Hyperparameters\n        batch_size = 64\n        epochs = 10\n        learning_rate = 0.01\n        momentum = 0.9\n\n        # Optimizer\n        optimizer = optim.SGD(model.parameters(), lr=learning_rate, momentum=momentum)\n\n        # Data loaders\n        transform = transforms.Compose([transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,))])\n        train_loader = DataLoader(datasets.MNIST('./data', train=True, download=True, transform=transform),\n                                  batch_size=batch_size, shuffle=True)\n        test_loader = DataLoader(datasets.MNIST('./data', train=False, transform=transform),\n                                 batch_size=batch_size, shuffle=False)\n\n        for epoch in range(1, epochs + 1):\n            train(model, device, train_loader, optimizer, epoch)\n            evaluate(model, device, test_loader)\n\n        # Save the trained model\n        torch.save(model.state_dict(), model_path)\n\n    # Predict and create submission file\n    test_csv_path = \"/kaggle/input/digit-recognizer/test.csv\"  # Path to the test CSV file\n    submission_csv_path = \"./submission.csv\"  # Path to the submission CSV file\n    predict_and_create_submission(model, test_csv_path, submission_csv_path)\n\n","metadata":{"execution":{"iopub.status.busy":"2024-06-21T19:55:31.471603Z","iopub.execute_input":"2024-06-21T19:55:31.47238Z","iopub.status.idle":"2024-06-21T19:55:31.482373Z","shell.execute_reply.started":"2024-06-21T19:55:31.472354Z","shell.execute_reply":"2024-06-21T19:55:31.481478Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"main()","metadata":{"execution":{"iopub.status.busy":"2024-06-21T19:55:31.48347Z","iopub.execute_input":"2024-06-21T19:55:31.483729Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{},"execution_count":null,"outputs":[]}]}