{"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, Model\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 = 192\nEPOCHS = 100\nINITIAL_LR = 1.5e-4\nEND_LR = 6e-6\nWEIGHT_DECAY = 2.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 bn_relu_conv(x, growth_rate, kernel_size=3):\n        x = layers.BatchNormalization()(x)\n        x = layers.ReLU()(x)\n        x = layers.Conv2D(growth_rate, kernel_size, padding='same')(x)\n        return x\n\n    def bottleneck_layer(x, growth_rate):\n        inter_channel = 4 * growth_rate\n        x1 = bn_relu_conv(x, inter_channel, kernel_size=1)\n        x1 = bn_relu_conv(x1, growth_rate, kernel_size=3)\n        x = layers.Concatenate(axis=-1)([x, x1])\n        x = layers.Dropout(DROPOUT_RATE / 2)(x)\n        return x\n\n    input_tensor = layers.Input(shape=(IMAGE_WIDTH, IMAGE_HEIGHT, 1))\n    x = layers.Conv2D(48, (5, 5), strides=2, padding='same')(input_tensor)\n    x = layers.BatchNormalization()(x)\n    x = layers.ReLU()(x)\n\n    for _ in range(3):\n        x = bottleneck_layer(x, 96)\n\n    x = layers.Conv2D(96, (5, 5), strides=2, padding='same')(x)\n    x = layers.BatchNormalization()(x)\n    x = layers.ReLU()(x)\n    x = layers.Dropout(DROPOUT_RATE / 2)(x)\n\n    x = layers.Flatten()(x)\n    \n    x = layers.Dense(384, activation='relu')(x)\n    x = layers.Dropout(DROPOUT_RATE)(x)\n    x = layers.Dense(192, activation='relu')(x)\n    x = layers.Dropout(DROPOUT_RATE / 2)(x)\n    x = layers.Dense(10, activation='softmax')(x)\n\n    model = Model(inputs=input_tensor, outputs=x)\n\n    print(\"Model depth:\", 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","trusted":true},"execution_count":null,"outputs":[]}]}