image_superresolution_final.py

# -*- coding: utf-8 -*-
"""Image_superresolution final

Automatically generated by Colab.

Original file is located at
    https://colab.research.google.com/#fileId=https%3A//storage.googleapis.com/kaggle-colab-exported-notebooks/image-superresolution-final-889a23da-1400-42f7-80b0-fcaca9e2df7a.ipynb%3FX-Goog-Algorithm%3DGOOG4-RSA-SHA256%26X-Goog-Credential%3Dgcp-kaggle-com%2540kaggle-161607.iam.gserviceaccount.com/20241002/auto/storage/goog4_request%26X-Goog-Date%3D20241002T050323Z%26X-Goog-Expires%3D259200%26X-Goog-SignedHeaders%3Dhost%26X-Goog-Signature%3D779bc4d9ff67819a045488e48069519c0079b12299c04d102d0bc1535a111dcea50ea023d703ea6c8bed50cf03493c7f5910bacba4a0cd5eef00a57ea5e0f36902a1e29520fb1a9befc686703658ce1ac8b9c6b0e7a119a13293469b10ed24a11bd61c5675b1d5d1e46bc328e558862d476e8eb638bf338c442b8294df14d06dae9ebdc948183e1f97ce8d007799ca98dc8f78a8e6513caa31cd7df4d748500ada48b432a42476539d7546514f8f112d730c2a3c3cce8a5abf4ef307702e9f3872392bf3803e5c4c79e1eba99fcfec77107f11753b8e0cf7ed442381e36a211132e51bb1c4c9e3fa493c04118c566447758a0eb9f1c231966ad52e44b297251a
"""

# IMPORTANT: RUN THIS CELL IN ORDER TO IMPORT YOUR KAGGLE DATA SOURCES
# TO THE CORRECT LOCATION (/kaggle/input) IN YOUR NOTEBOOK,
# THEN FEEL FREE TO DELETE THIS CELL.
# NOTE: THIS NOTEBOOK ENVIRONMENT DIFFERS FROM KAGGLE'S PYTHON
# ENVIRONMENT SO THERE MAY BE MISSING LIBRARIES USED BY YOUR
# NOTEBOOK.

import os
import sys
from tempfile import NamedTemporaryFile
from urllib.request import urlopen
from urllib.parse import unquote, urlparse
from urllib.error import HTTPError
from zipfile import ZipFile
import tarfile
import shutil

CHUNK_SIZE = 40960
DATA_SOURCE_MAPPING = 'image-super-resolution:https%3A%2F%2Fstorage.googleapis.com%2Fkaggle-data-sets%2F810739%2F1388983%2Fbundle%2Farchive.zip%3FX-Goog-Algorithm%3DGOOG4-RSA-SHA256%26X-Goog-Credential%3Dgcp-kaggle-com%2540kaggle-161607.iam.gserviceaccount.com%252F20241002%252Fauto%252Fstorage%252Fgoog4_request%26X-Goog-Date%3D20241002T050323Z%26X-Goog-Expires%3D259200%26X-Goog-SignedHeaders%3Dhost%26X-Goog-Signature%3D0a0f21c217092ec421a67e299c5622a4568678393d155e5494088dc840066e7e7900add726e92334d4d0be06f1efafcbd2aab930e1fa45e45e7ca8558a2faab35ed806c1779680a570cac97931d718ad064b109f98ac87b3e918676d3227f4d3347c33f116c8ef41d3b31d791db3718aceb43d3a285182d9721ae5b10bfb8a4cdfda3c6de83372bb0d3056b674073db184168f8c091b7f58b98d09da5acd2b673867ca406b3d2fdeaa281c6a498d750394346717018dd12b6c103412080523e6000cff7d18d77bf4a41e142a8c6386732335ebb094b080178dd712f62201bb3710c7ec1663c60e32e8be3f134c353d81c369a1d72cc2402df762e6ceaf54b98e'

KAGGLE_INPUT_PATH='/kaggle/input'
KAGGLE_WORKING_PATH='/kaggle/working'
KAGGLE_SYMLINK='kaggle'

!umount /kaggle/input/ 2> /dev/null
shutil.rmtree('/kaggle/input', ignore_errors=True)
os.makedirs(KAGGLE_INPUT_PATH, 0o777, exist_ok=True)
os.makedirs(KAGGLE_WORKING_PATH, 0o777, exist_ok=True)

try:
  os.symlink(KAGGLE_INPUT_PATH, os.path.join("..", 'input'), target_is_directory=True)
except FileExistsError:
  pass
try:
  os.symlink(KAGGLE_WORKING_PATH, os.path.join("..", 'working'), target_is_directory=True)
except FileExistsError:
  pass

for data_source_mapping in DATA_SOURCE_MAPPING.split(','):
    directory, download_url_encoded = data_source_mapping.split(':')
    download_url = unquote(download_url_encoded)
    filename = urlparse(download_url).path
    destination_path = os.path.join(KAGGLE_INPUT_PATH, directory)
    try:
        with urlopen(download_url) as fileres, NamedTemporaryFile() as tfile:
            total_length = fileres.headers['content-length']
            print(f'Downloading {directory}, {total_length} bytes compressed')
            dl = 0
            data = fileres.read(CHUNK_SIZE)
            while len(data) > 0:
                dl += len(data)
                tfile.write(data)
                done = int(50 * dl / int(total_length))
                sys.stdout.write(f"\r[{'=' * done}{' ' * (50-done)}] {dl} bytes downloaded")
                sys.stdout.flush()
                data = fileres.read(CHUNK_SIZE)
            if filename.endswith('.zip'):
              with ZipFile(tfile) as zfile:
                zfile.extractall(destination_path)
            else:
              with tarfile.open(tfile.name) as tarfile:
                tarfile.extractall(destination_path)
            print(f'\nDownloaded and uncompressed: {directory}')
    except HTTPError as e:
        print(f'Failed to load (likely expired) {download_url} to path {destination_path}')
        continue
    except OSError as e:
        print(f'Failed to load {download_url} to path {destination_path}')
        continue

print('Data source import complete.')

# This Python 3 environment comes with many helpful analytics libraries installed
# It is defined by the kaggle/python Docker image: https://github.com/kaggle/docker-python
# For example, here's several helpful packages to load

import numpy as np # linear algebra
import pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)

# Input data files are available in the read-only "../input/" directory
# For example, running this (by clicking run or pressing Shift+Enter) will list all files under the input directory

import os
for dirname, _, filenames in os.walk('/kaggle/input'):
    for filename in filenames:
        print(os.path.join(dirname, filename))

# 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"
# You can also write temporary files to /kaggle/temp/, but they won't be saved outside of the current session

import tensorflow as tf
from tensorflow.keras import layers, models, Input
from tensorflow.keras.callbacks import EarlyStopping

# Residual Unit with two Conv layers and a residual connection
def residual_unit(input_tensor, filters, kernel_size=3):
    x = layers.Conv2D(filters, (kernel_size, kernel_size), padding='same')(input_tensor)
    x = layers.Activation('relu')(x)
    x = layers.Conv2D(filters, (kernel_size, kernel_size), padding='same')(x)
    return layers.Add()([input_tensor, x])  # Local residual connection

# Recursive Block (with shared weights among residual units)
def recursive_block(input_tensor, num_residual_units, filters):
    x = input_tensor
    for _ in range(num_residual_units):
        x = residual_unit(x, filters)
    return x

# DRRN Model
def DRRN(input_shape, num_recursive_blocks, num_residual_units, filters):
    inputs = Input(shape=input_shape)

    # Initial Conv Layer
    x = layers.Conv2D(filters, (3, 3), padding='same')(inputs)

    # Recursive blocks with residual connections
    for _ in range(num_recursive_blocks):
        x = recursive_block(x, num_residual_units, filters)

    # Output Conv Layer to reconstruct residual image
    x = layers.Conv2D(3, (3, 3), padding='same')(x)

    # Global Residual Learning (add the input LR image to the predicted residual)
    outputs = layers.Add()([inputs, x])

    # Build model
    model = models.Model(inputs, outputs)
    return model

# Define the input shape (LR image), number of recursive blocks, residual units, and filters
input_shape = (31, 31, 3)  # (height, width, channels)
num_recursive_blocks = 3  # Number of recursive blocks
num_residual_units = 9    # Number of residual units in each block
filters = 64              # Number of filters in Conv layers

# Build the DRRN model
drrn_model = DRRN(input_shape, num_recursive_blocks, num_residual_units, filters)

# Gradient Clipping with Adam optimizer
optimizer = tf.keras.optimizers.Adam(
    learning_rate=0.001,  # You can adjust this learning rate
    clipnorm=1.0  # This clips gradients to a maximum L2 norm of 1.0
)

# Compile the model with gradient clipping
drrn_model.compile(optimizer=optimizer, loss='mse', metrics=['accuracy'])

# Print the model summary
drrn_model.summary()

import numpy as np
import os
import cv2

# Function to extract 31x31 patches with a stride of 21 from a single image
def extract_patches(image, patch_size=31, stride=21):
    patches = []
    h, w, _ = image.shape
    for y in range(0, h - patch_size + 1, stride):
        for x in range(0, w - patch_size + 1, stride):
            patch = image[y:y + patch_size, x:x + patch_size]
            patches.append(patch)
    return np.array(patches)

# Function to load images from a directory and extract low-res and high-res patches
def load_and_extract_patches(low_res_dir, high_res_dir, patch_size=31, stride=21):
    low_res_patches = []
    high_res_patches = []

    # Assuming filenames are identical for both low-res and high-res images
    for file_name in os.listdir(low_res_dir):
        # Load low-resolution image
        low_res_image_path = os.path.join(low_res_dir, file_name)
        low_res_image = cv2.imread(low_res_image_path)

        # Load high-resolution image
        high_res_image_path = os.path.join(high_res_dir, file_name)
        high_res_image = cv2.imread(high_res_image_path)

        # Ensure both images have at least patch_size dimensions and are the same size
        if (low_res_image.shape[0] >= patch_size and low_res_image.shape[1] >= patch_size and
            high_res_image.shape == low_res_image.shape):

            # Extract patches from both low-res and high-res images
            low_res_patches.extend(extract_patches(low_res_image, patch_size, stride))
            high_res_patches.extend(extract_patches(high_res_image, patch_size, stride))

    return np.array(low_res_patches), np.array(high_res_patches)

# Define the paths to your low-resolution and high-resolution images
low_res_dir = "/kaggle/input/image-super-resolution/dataset/Raw Data/low_res"
high_res_dir = "/kaggle/input/image-super-resolution/dataset/Raw Data/high_res"

# Extract low-res and high-res patches
X_train, Y_train = load_and_extract_patches(low_res_dir, high_res_dir, patch_size=31, stride=21)

# Normalize the patches (if necessary)
X_train = X_train / 255.0  # Normalize low-res patches (input)
Y_train = Y_train / 255.0  # Normalize high-res patches (target)

# Reshape the data for model input
X_train = X_train.reshape(-1, 31, 31, 3)  # Low-res patches
Y_train = Y_train.reshape(-1, 31, 31, 3)  # High-res patches

print(f"Extracted {X_train.shape[0]} patches of size 31x31 from the dataset.")

early_stopping = EarlyStopping(
    monitor='val_loss',  # Monitors validation loss
    patience=5,          # Number of epochs with no improvement after which training will be stopped
    restore_best_weights=True  # Restores model weights from the epoch with the best validation loss
)

history = drrn_model.fit(
    X_train, Y_train,
    epochs=50,
    batch_size=128,
    validation_split=0.1,
    callbacks=[early_stopping]  # Include the early stopping callback
)

# Save the trained model
drrn_model.save("drrn_super_resolution.h5")

import random
import matplotlib.pyplot as plt
def show_random_patch(X_train, Y_train, model, num_patches=5):
    # Select random indices from the dataset
    indices = random.sample(range(X_train.shape[0]), num_patches)

    for idx in indices:
        # Get the low-quality patch (input)
        low_quality_patch = X_train[idx]

        # Get the high-quality patch (ground truth)
        high_quality_patch = Y_train[idx]

        # Predict the improved patch using the DRRN model
        low_quality_patch_input = low_quality_patch[np.newaxis, ...]  # Add batch dimension
        improved_patch = model.predict(low_quality_patch_input)[0]  # Remove batch dimension

        # Plot the low-quality, improved, and high-quality patches side-by-side
        fig, axes = plt.subplots(1, 3, figsize=(12, 4))
        axes[0].imshow(low_quality_patch)
        axes[0].set_title("Low-Quality Patch")
        axes[0].axis('off')

        axes[1].imshow(improved_patch)
        axes[1].set_title("Improved Patch (Predicted)")
        axes[1].axis('off')

        axes[2].imshow(high_quality_patch)
        axes[2].set_title("High-Quality Patch (Ground Truth)")
        axes[2].axis('off')

        plt.show()

# Example usage: show 5 random patches from the dataset
show_random_patch(X_train, Y_train, drrn_model, num_patches=5)