{"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"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import os\nimport numpy as np\nimport pandas as pd\nimport tensorflow as tf\nfrom tensorflow.keras import layers\nfrom tensorflow.keras.models import load_model\nfrom tensorflow.keras.callbacks import ModelCheckpoint, LearningRateScheduler\nfrom tensorflow.keras.preprocessing.image import ImageDataGenerator\nimport tensorflow_addons as tfa\n\n# Constants\nTRAIN_CSV = \"/kaggle/input/digit-recognizer/train.csv\"\nTEST_CSV = \"/kaggle/input/digit-recognizer/test.csv\"\nSUBMISSION_FILE = \"/kaggle/working/submission.csv\"\nMODEL_PATH = \"/kaggle/working/best_model.h5\"\nIMAGE_WIDTH, IMAGE_HEIGHT = 28, 28\nBATCH_SIZE = 24\nEPOCHS = 100\nINITIAL_LR = 1.5e-4\nEND_LR = 6e-6\nWEIGHT_DECAY = 5e-6\nDROPOUT_RATE = 0.5\n\n# Config\nUSE_VALIDATION_SPLIT = True\n\nclass PolynomialWithWarmupAndFineTuningSchedule(LearningRateScheduler):\n    def __init__(self, initial_lr, end_lr, total_epochs, warmup_epochs=10, fine_tuning_epochs=2, power=1.0):\n        self.initial_lr = initial_lr\n        self.end_lr = end_lr\n        self.total_epochs = total_epochs\n        self.warmup_epochs = warmup_epochs\n        self.fine_tuning_epochs = fine_tuning_epochs\n        self.decay_epochs = self.total_epochs - self.warmup_epochs - self.fine_tuning_epochs\n        self.power = power\n\n    def __call__(self, epoch):\n        # Warmup phase with a gradual increase in learning rate\n        if epoch < self.warmup_epochs:\n            warmup_lr = self.initial_lr\n            return warmup_lr\n        # Decay phase\n        elif epoch < (self.total_epochs - self.fine_tuning_epochs):\n            decay_step = epoch - self.warmup_epochs\n            return self.end_lr + (self.initial_lr - self.end_lr) * (1 - decay_step / self.decay_epochs) ** self.power\n        # Fine-tuning phase with a constant learning rate\n        else:\n            return self.end_lr\n        \ndef load_data_from_csv(file_path, is_train=True):\n    df = pd.read_csv(file_path)\n    if is_train:\n        images = df.iloc[:, 1:].values.reshape(-1, IMAGE_WIDTH, IMAGE_HEIGHT, 1).astype(np.float32) / 255.0\n        labels = pd.get_dummies(df['label']).values\n        return images, labels\n    else:\n        images = df.values.reshape(-1, IMAGE_WIDTH, IMAGE_HEIGHT, 1).astype(np.float32) / 255.0\n        return images, None\n        \ndef create_datagen(validation_split=None):\n    \"\"\"Returns ImageDataGenerator with optional validation split.\"\"\"\n    kwargs = {\n        'rescale': 1./255,\n        'width_shift_range': 0.15,\n        'height_shift_range': 0.15,\n        'shear_range': 0.15,\n        'zoom_range': 0.15,\n        'rotation_range': 20, \n        'fill_mode': 'nearest'\n    }\n\n    if validation_split:\n        kwargs['validation_split'] = validation_split\n\n    return ImageDataGenerator(**kwargs)\n\ndef create_generators():\n    train_images, train_labels = load_data_from_csv(TRAIN_CSV, is_train=True)\n    test_images, _ = load_data_from_csv(TEST_CSV, is_train=False)\n\n    # Data augmentation\n    datagen_args = {\n        'width_shift_range': 0.15,\n        'height_shift_range': 0.15,\n        'shear_range': 0.15,\n        'zoom_range': 0.15,\n        'rotation_range': 20,\n        'fill_mode': 'nearest'\n    }\n    \n    if USE_VALIDATION_SPLIT:\n        datagen_args['validation_split'] = 0.025\n\n    datagen = ImageDataGenerator(**datagen_args)\n    \n    train_generator_args = {\n        'x': train_images,\n        'y': train_labels,\n        'batch_size': BATCH_SIZE,\n        'shuffle': True\n    }\n\n    if USE_VALIDATION_SPLIT:\n        train_generator_args['subset'] = 'training'\n\n    train_generator = datagen.flow(**train_generator_args)\n\n    validation_generator = None\n    if USE_VALIDATION_SPLIT:\n        validation_generator = datagen.flow(\n            x=train_images,\n            y=train_labels,\n            batch_size=BATCH_SIZE,\n            subset='validation'\n        )\n\n    return train_generator, validation_generator, test_images\n\ndef create_model():\n\n    def count_conv2d_layers(model):\n        conv_layer_types = (tf.keras.layers.Conv2D, tf.keras.layers.DepthwiseConv2D)\n        return sum([1 for layer in model.layers if isinstance(layer, conv_layer_types)])\n\n    def MBConv(inputs, out_channels, kernel_size, stride, expansion_factor, se_ratio=0.25, activation=tf.nn.silu):\n        # Expansion phase\n        x = layers.Conv2D(inputs.shape[-1] * expansion_factor, 1, padding='same', use_bias=False)(inputs)\n        x = layers.BatchNormalization()(x)\n        x = activation(x)\n        x = layers.Dropout(DROPOUT_RATE / 2)(x)\n\n        # Depthwise convolution\n        x = layers.DepthwiseConv2D(kernel_size, strides=stride, padding='same', use_bias=False)(x)\n        x = layers.BatchNormalization()(x)\n        x = activation(x)\n        x = layers.Dropout(DROPOUT_RATE / 2)(x)\n\n        # Squeeze and Excitation\n        se = layers.GlobalAveragePooling2D()(x)\n        se = layers.Reshape((1, 1, x.shape[-1]))(se)\n        squeezed = layers.Dense(x.shape[-1] * se_ratio, activation=activation)(se)\n        excited = layers.Dense(x.shape[-1], activation='sigmoid')(squeezed)\n        x = layers.Multiply()([x, excited])\n\n        # Output phase\n        x = layers.Conv2D(out_channels, 1, padding='same', use_bias=False)(x)\n        x = layers.BatchNormalization()(x)\n        return x\n    \n    input_tensor = layers.Input(shape=(IMAGE_WIDTH, IMAGE_HEIGHT, 1))\n\n    x = layers.Conv2D(384, 3, strides=2, padding='same')(input_tensor)\n    x = layers.BatchNormalization()(x)\n    x = tf.nn.silu(x)\n    x = layers.Dropout(DROPOUT_RATE / 2)(x)\n\n    # MBConv(x, channels, kernel_size, stride, expansion_factor): Applies an MBConv block with specified channels, kernel size, stride, and expansion factor to input x.\n    x = MBConv(x, 384, 3, 1, 2)\n    x = MBConv(x, 384, 3, 1, 2)\n    \n    x = MBConv(x, 768, 4, 1, 4)\n    \n    x = layers.Conv2D(768, 4, strides=1, padding='same')(x)\n    x = layers.BatchNormalization()(x)\n    x = tf.nn.silu(x)\n    x = layers.Dropout(DROPOUT_RATE / 2)(x)\n    \n    x = layers.Conv2D(768, 4, strides=1, padding='same')(x)\n    x = layers.BatchNormalization()(x)\n    x = tf.nn.silu(x)\n    x = layers.Dropout(DROPOUT_RATE / 2)(x)\n    \n    x = layers.Conv2D(768, 5, strides=1, padding='valid')(x)\n    x = layers.BatchNormalization()(x)\n    x = tf.nn.silu(x)\n    x = layers.Dropout(DROPOUT_RATE / 2)(x)\n    \n    x = layers.Conv2D(768, 5, strides=1, padding='valid')(x)\n    x = layers.BatchNormalization()(x)\n    x = tf.nn.silu(x)\n    x = layers.Dropout(DROPOUT_RATE / 2)(x)\n\n    # Flatten layer to reshape the tensor for dense layers\n    x = layers.Flatten()(x)\n    \n    # Top layers\n    x = layers.Dense(512, activation=\"relu\")(x)\n    x = layers.Dropout(DROPOUT_RATE)(x)\n    x = layers.Dense(384, activation=\"relu\")(x)\n    x = layers.Dropout(DROPOUT_RATE)(x)\n    x = layers.Dense(10, activation=\"softmax\")(x)\n\n    model = tf.keras.models.Model(inputs=input_tensor, outputs=x)\n\n    print(\"Model dept:\", count_conv2d_layers(model))\n    print(\"Model Output Shape:\", model.output_shape)\n\n    return model\n\ndef main():\n\n    train_generator, validation_generator, test_images = create_generators()\n    model = create_model()\n\n    scheduler = PolynomialWithWarmupAndFineTuningSchedule(INITIAL_LR, END_LR, EPOCHS)\n    lr_callback = LearningRateScheduler(scheduler)\n    optimizer = tfa.optimizers.AdamW(learning_rate=INITIAL_LR, weight_decay=WEIGHT_DECAY)\n    \n    model.compile(optimizer=optimizer, loss='categorical_crossentropy', metrics=['accuracy'])\n\n    model_checkpoint = ModelCheckpoint(\n        MODEL_PATH,\n        monitor='val_loss',\n        save_best_only=True,\n        mode='min',\n        verbose=1\n    )\n\n    model.fit(\n        train_generator,\n        epochs=EPOCHS,\n        validation_data=validation_generator if USE_VALIDATION_SPLIT else None,\n        callbacks=[lr_callback, model_checkpoint]\n    )\n\n    best_model = load_model(MODEL_PATH)\n    predictions = best_model.predict(test_images, batch_size=BATCH_SIZE)\n    class_predictions = np.argmax(predictions, axis=1)\n    submission = pd.DataFrame({\"ImageId\": range(1, len(class_predictions) + 1), \"Label\": class_predictions})\n    submission.to_csv(SUBMISSION_FILE, index=False)\n\nif __name__ == \"__main__\":\n    main()","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","execution":{"iopub.status.busy":"2023-09-18T22:16:28.730991Z","iopub.execute_input":"2023-09-18T22:16:28.731282Z"},"trusted":true},"execution_count":null,"outputs":[]}]}