Upload utils.py

utils.py

@@ -0,0 +1,69 @@
+# ==============================================
+# utils.py  |  Helper functions for Super-Resolution
+# ==============================================
+
+%%writefile utils.py
+import tensorflow as tf
+import tensorflow_datasets as tfds
+
+# --- Define Image Dimensions ---
+HR_SIZE = (128, 128)
+LR_SIZE = (HR_SIZE[0] // 2, HR_SIZE[1] // 2)  # 64x64
+
+
+def preprocess_image(data):
+    """
+    Normalizes and resizes images for the dataset.
+    The model's input will be a bicubic-upscaled version of the LR image.
+    """
+    # Normalize pixel values to [0, 1]
+    hr_image = tf.cast(data['hr'], tf.float32) / 255.0
+    lr_image = tf.cast(data['lr'], tf.float32) / 255.0
+
+    # Resize to target dimensions
+    hr_image = tf.image.resize(hr_image, HR_SIZE, method='bicubic')
+    lr_image = tf.image.resize(lr_image, LR_SIZE, method='bicubic')
+
+    # Create model input by upscaling the low-res image
+    model_input_image = tf.image.resize(lr_image, HR_SIZE, method='bicubic')
+
+    return model_input_image, hr_image
+
+
+def load_div2k_data(batch_size=16):
+    """
+    Loads the DIV2K dataset and creates an efficient tf.data pipeline.
+    Uses the 'bicubic_x2' version for 2x super-resolution.
+    """
+    print("Loading and preparing DIV2K dataset...")
+
+    # Load the dataset using TensorFlow Datasets
+    (train_ds, valid_ds), ds_info = tfds.load(
+        'div2k/bicubic_x2',
+        split=['train', 'validation'],
+        as_supervised=False,  # We provide our own preprocessing
+        with_info=True
+    )
+
+    # --- Create the Training Pipeline ---
+    train_dataset = (
+        train_ds
+        .map(preprocess_image, num_parallel_calls=tf.data.AUTOTUNE)
+        .cache()                     # Cache for performance
+        .shuffle(buffer_size=100)
+        .batch(batch_size)
+        .repeat()                    # Repeat for multiple epochs
+        .prefetch(buffer_size=tf.data.AUTOTUNE)
+    )
+
+    # --- Create the Validation Pipeline ---
+    validation_dataset = (
+        valid_ds
+        .map(preprocess_image, num_parallel_calls=tf.data.AUTOTUNE)
+        .batch(batch_size)
+        .cache()
+        .prefetch(buffer_size=tf.data.AUTOTUNE)
+    )
+
+    print("✅ Dataset pipelines created successfully.")
+    return train_dataset, validation_dataset, ds_info