1. Context and motivation: Why should you care about this notebook?
\n
\n\n[Back to table of contents](#TOC)\n\nHere is why:\n\n\n\nThe **robustness** of the models we build has always been an issue:\n- **Natural robustness** means that your model is very likely to make good decisions in \"naturel\" scenarios, i.e. contexts that you would expect to encounter in your everyday life. For example, an animal classifier expects to be given animal pictures as inputs.\n- **Adversarial robustness** means that your model is not vulnerable to *evasion attacks*, i.e. cyberattacks that will slightly change the input image in such way that a near-perfect classifier would fail miserably at each input. How? By adding an imperceptible noise, carefully crafted based on knowledge extracted by querying the model multiple times (see example above, where an XGBoost model is fooled by only changing a single pixel of the original, correctly classified image). Evasion attacks have proven to be very effective on image datasets and consistute obvious security threats.\n\nTherefore, *how should you make your models safe?* One way of doing that is **data augmentation**. This means that you will craft new images for your dataset based on the existing ones, in order to improve the versatility of the dataset and force the trained model to learn more general patterns rather that simplistic ones. Data augmentation considerably rises the robustness of the trained model, although the training time is also longer.\n\n
\n Remark: So far, this package can only treat and export jpg images. If your images are png, please consider building a simple loop to change the extensions into jpg. If this is really an issue for you, you can contact me for an update of the fast_skimage library.\n
2. Data augmentation techniques available in this library
\n
\n\n[Back to table of contents](#TOC)\n\n\n### ✅ **1. Random Rotation**\n**Function:** `random_rotate`\n\nRandomly rotates the image within a user-defined angle range (by default, ±25 degrees). Rotation helps your model become invariant to orientation changes, making it robust to rotated images encountered in real-world scenarios.\n\n\n### ✅ **2. Random Shear**\n**Function:** `random_shear`\n\nRandomly shears (slants) your image by a small angle. This augmentation helps the model to generalize better to slight geometric distortions or perspective changes in real-world images.\n\n\n### ✅ **3. Random Translation**\n**Function:** `random_translate`\n\nRandomly shifts the image horizontally and vertically within user-specified pixel limits. Translation forces your model to become insensitive to the exact placement of the object within the image frame, improving generalization.\n\n\n### ✅ **4. Random Reflection (Flip)**\n**Function:** `random_reflect`\n\nRandomly flips your image either vertically, horizontally, or both. Reflection helps build invariance towards mirrored images and is particularly effective for datasets where orientation does not matter (such as certain natural scenes or objects).\n\n\n### ✅ **5. Random Scale (Zoom)**\n**Function:** `random_scale`\n\nRandomly zooms in or out by a small factor. Scaling simulates objects being observed from different distances, enhancing the model's ability to handle varied object sizes.\n\n\n### ✅ **6. Spatial Padding**\n**Function:** `spatial_pad`\n\nPads the borders of your image with zeros or a mean pixel value to reach a target dimension. Padding is especially useful in convolutional neural networks (CNNs) to preserve the input size after multiple layers of convolutions, helping the model learn stable patterns across image boundaries.\n\n\n### ✅ **7. Ben Graham Transform (APTOS method)**\n**Function:** `ben_graham_transform`\n\nApplies an effective preprocessing step inspired by Ben Graham's winning solution in the APTOS competition (see [here](https://www.kaggle.com/code/nxrprime/panda-image-data-augmentation-techniques-w-monai)). This method emphasizes image features by adding a weighted version of a Gaussian-blurred copy of the image back onto itself. It's particularly powerful for enhancing image contrast and structural clarity, benefiting medical and photographic images alike.\n\n\n### ✅ **8. Random Gaussian Noise**\n**Function:** `random_gaussian_noise`\n\nInjects random Gaussian noise into the image. Adding Gaussian noise emulates real-life sensor noise or environmental variations, making your model robust to noisy input data and less sensitive to small perturbations (important for adversarial robustness).\n\n\n### ✅ **9. Random Salt-and-Pepper Noise**\n**Function:** `random_salt_pepper_noise`\n\nRandomly applies \"salt-and-pepper\" (black-and-white) pixel noise to the image. This augmentation simulates potential corruption or pixel-level faults, training the model to ignore spurious artifacts and handle corrupted data gracefully.\n\n\n### ✅ **10. Random Text Overlay**\n**Function:** `random_text_overlay`\n\nAdds random, semi-transparent text at random positions on your image. This augmentation improves robustness towards textual artifacts or watermarks that might appear in real-world image sources. It's especially useful when models must ignore irrelevant textual content on images.\n\n\n
\nHow to use these augmentations:\n\nSimply create a list of the desired augmentation functions and pass it to the augmentation method in the provided library. You can control the number of augmentations applied randomly per image, ensuring a diverse augmented dataset.\n
\n\nHere's an example of how to define and use them:\n\n```python\naugmentations = [\n random_rotate, \n random_shear, \n random_translate, \n random_reflect, \n random_scale, \n spatial_pad, \n ben_graham_transform, \n random_gaussian_noise,\n random_salt_pepper_noise,\n random_text_overlay\n]\n\n# Example usage:\naug.build_augmented_dataset_on_disk(\n dataset_id=0,\n output_dir=\"augmented_dataset\",\n scale_factor=2, # generates 2 new images per original image\n augmentations=augmentations,\n min_augmentations=1, # At least 1 augmentation technique will be applied on each image\n max_augmentations=5, # No image can contain more than 5 augmentations simultaneously\n label=\"cat\",\n to_df=True,\n csv_name=\"augmented_dataset.csv\"\n)\n\n","metadata":{}},{"cell_type":"markdown","source":"\n
\n
3. Environment setup
\n
\n\n[Back to table of contents](#TOC)","metadata":{}},{"cell_type":"code","source":"#| default_exp core","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-03-24T18:39:11.616754Z","iopub.execute_input":"2025-03-24T18:39:11.617048Z","iopub.status.idle":"2025-03-24T18:39:11.620886Z","shell.execute_reply.started":"2025-03-24T18:39:11.617015Z","shell.execute_reply":"2025-03-24T18:39:11.619973Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#| export\n\ndef ensure_dependencies_installed():\n \"\"\"\n Ensures required dependencies are installed.\n \"\"\"\n try:\n import monai, tqdm, cv2, fast_skimage, skimage, tensorflow, numpy\n except ImportError as e:\n print(\"Missing libraries detected. You can download them by runing 'Add-ons/Install Dependencies' on Kaggle or simply uncommenting the line below.\")\n # !pip install monai tqdm opencv-python fast-skimage scikit-image tensorflow numpy\n\n# Call to make sure libraries are installed\nensure_dependencies_installed()\n\n# Global imports\nimport os\nimport warnings\nimport shutil\nimport sys\nimport random\nfrom skimage.io import imsave\nimport pkg_resources # Track lib versions\nimport gc # Garbage collector\nimport glob\nimport pandas as pd # Operations on CSV files\nimport cv2 # Operations on images\nimport numpy as np\nimport fast_skimage\nfrom skimage.transform import rotate\nfrom skimage.transform import AffineTransform, warp\nfrom scipy.ndimage import gaussian_filter\nfrom fast_skimage.Image import Image\nfrom fast_skimage.Collection import Collection\nimport torch\nfrom PIL import Image as PILImage, ImageDraw, ImageFont\nimport string\nimport matplotlib.pyplot as plt\n\nimport tensorflow as tf\nfrom tensorflow.keras.preprocessing.image import load_img, img_to_array\nfrom tensorflow.keras.models import Sequential\nfrom tensorflow.keras.layers import Input, Lambda, Flatten, Dropout, Dense\nfrom tensorflow.keras.optimizers import Adamax\nfrom tensorflow.keras.metrics import Precision, Recall\nfrom tensorflow.keras.applications import Xception\nfrom sklearn.model_selection import train_test_split\n\nfrom monai.transforms import (\n LoadImage,\n RandAffine,\n Rand2DElastic,\n Rand3DElastic,\n RandFlip,\n SpatialPad,\n Transform,\n Compose\n)\n\nfrom tqdm.notebook import tqdm\n\nwarnings.filterwarnings(\"ignore\")\n\n# Set random seed\nRANDOM_SEED = 5\ntf.random.set_seed(RANDOM_SEED)\nnp.random.seed(RANDOM_SEED)\n\nDEFAULT_OUTPUT_DIR = \"Augmented data\"\n# !mkdir \"Augmented data\"\n\n# Display versions (no re-import necessary)\nprint(f\"Python version: {sys.version}\\n\")\n\nprint(\"Num GPUs Available: \", len(tf.config.list_physical_devices('GPU')), \"\\n\")\n\nprint(\"Installed libraries and versions:\")\ninstalled_packages = pkg_resources.working_set\npackages_list = sorted([f\"{i.key}=={i.version}\" for i in installed_packages])\n\nfor package in packages_list:\n print(package)\n\ngc.collect()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-03-24T18:39:11.644107Z","iopub.execute_input":"2025-03-24T18:39:11.644400Z","iopub.status.idle":"2025-03-24T18:39:40.486432Z","shell.execute_reply.started":"2025-03-24T18:39:11.644378Z","shell.execute_reply":"2025-03-24T18:39:40.485635Z"},"_kg_hide-output":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"\n
\n
4. Building a data augmentation class
\n
\n\n[Back to table of contents](#TOC)\n\nNow, we will build the `Augmenter` class in which operations on dataset will be executed. The idea is to input a dataset and list of parameters, and output the augmented dataset. Wraping this into a class makes it easier for you to reuse this in your own work via kagglehub.\n\nThe `Augmenter` class makes an extensive usage of the `Collection` class from `fast_skimage` (which I designed). I built this library to ease up image processing. Each element of a collection is stored in an optimized `Image` class that allows you to perform a series of processing operations very easily. Below is a simple example of how to handle the `Collection` class:\n\n```Python\nfor im in collection_cat.images.values():\n im.color_to_gray() # This made all cat images turn gray!\n\nfor im in collection_dog.images.values():\n collection_cat.images[len(collection_cat.images.values())+1] = im # We added all dog images to cat images!\n```\n\nThe `Collection` class is therefore very convenient for image data augmentation tasks.","metadata":{}},{"cell_type":"code","source":"#| export\n\ndef random_rotate(img_obj, angle_range=(-25, 25)):\n \"\"\"\n Rotates the image by a random angle (in degrees) within angle_range.\n Ensures the array is (H, W) or (H, W, 3) to avoid ValueError.\n \"\"\"\n\n arr = img_obj.image\n\n # 1) If we have a channel-first array (C, H, W), transpose to (H, W, C).\n # Heuristic: the first dim is small (1..4) and second dim is large => likely (C, H, W).\n if arr.ndim == 3 and arr.shape[0] <= 4 and arr.shape[1] > 10:\n arr = np.transpose(arr, (1, 2, 0))\n\n # 2) If an alpha channel is present (H, W, 4), drop it (we only need 3 channels).\n if arr.ndim == 3 and arr.shape[-1] == 4:\n arr = arr[..., :3]\n\n # 3) If single-channel last dim => squeeze out to (H, W).\n # i.e. shape (H, W, 1) => (H, W).\n if arr.ndim == 3 and arr.shape[-1] == 1:\n arr = arr[..., 0] # or np.squeeze(arr, axis=-1)\n\n # Overwrite the image in the Image object with our standardized array\n img_obj.image = arr\n\n # 4) Rotate by random angle\n angle = np.random.uniform(angle_range[0], angle_range[1])\n try:\n # rotate(...) returns float in [0..1], so preserve_range => keep original range\n rotated = rotate(img_obj.image, angle, preserve_range=True)\n rotated = np.clip(rotated, 0, 255).astype(img_obj.image.dtype)\n \n # 5) Update the Image object\n img_obj.set_image(rotated, method=f\"random_rotate_{round(angle,1)}deg\")\n except Exception as e:\n print(\"An error occured while applying rotation for an image:\", e)\n return img_obj\n\n\ndef random_text_overlay(img_obj, prob=0.8,\n font_size_range=(12, 40),\n opacity_range=(100, 255),\n num_words_range=(1, 3)):\n \"\"\"\n Overlay random text on the image with a given probability.\n\n Args:\n img_obj: Your image object (expects .image as NumPy array).\n prob: Probability of applying this augmentation.\n font_size_range: Tuple specifying the min/max font size.\n opacity_range: Tuple specifying the min/max opacity (alpha channel).\n num_words_range: Number of words in random text.\n\n Returns:\n img_obj with random text overlay.\n \"\"\"\n\n if random.random() > prob:\n return img_obj # No augmentation applied\n\n arr = img_obj.image\n\n # Convert to PIL image\n if arr.ndim == 2:\n mode = 'L' # grayscale\n elif arr.shape[2] == 3:\n mode = 'RGB'\n elif arr.shape[2] == 4:\n mode = 'RGBA'\n else:\n raise ValueError(\"Unexpected image shape for text overlay\")\n\n pil_img = PILImage.fromarray(arr.astype('uint8'), mode=mode).convert(\"RGBA\")\n\n # Create overlay for transparency\n txt_overlay = PILImage.new('RGBA', pil_img.size, (255, 255, 255, 0))\n draw = ImageDraw.Draw(txt_overlay)\n\n # Random text generation\n num_words = random.randint(*num_words_range)\n random_text = ' '.join(''.join(random.choices(string.ascii_letters, k=random.randint(3,8))) for _ in range(num_words))\n\n # Random font size\n font_size = random.randint(*font_size_range)\n \n try:\n font = ImageFont.truetype(\"arial.ttf\", font_size)\n except:\n font = ImageFont.load_default()\n\n # Use textbbox to measure text size instead of deprecated textsize\n text_bbox = draw.textbbox((0, 0), random_text, font=font)\n text_width = text_bbox[2] - text_bbox[0]\n text_height = text_bbox[3] - text_bbox[1]\n\n # Text position\n width, height = pil_img.size\n position = (random.randint(0, max(0, width - text_width)),\n random.randint(0, max(0, height - text_height)))\n\n # Random text color\n text_color = tuple(np.random.randint(0, 256, size=3))\n\n # Random opacity\n opacity = random.randint(*opacity_range)\n\n # Draw the text\n draw.text(position, random_text, font=font, fill=text_color + (opacity,))\n\n # Composite the image with overlay\n combined = PILImage.alpha_composite(pil_img, txt_overlay).convert(mode)\n\n # Back to NumPy array\n img_obj.set_image(np.array(combined), method=\"random_text_overlay\")\n\n return img_obj\n \n\ndef random_elastic_deformation(img_obj):\n arr = img_obj.image\n\n # Channel-first conversion\n if arr.ndim == 2:\n arr = arr[None, ...] # (1,H,W)\n elif arr.ndim == 3 and arr.shape[-1] == 3:\n arr = np.transpose(arr, (2, 0, 1)) # (3,H,W)\n\n # Add dummy third spatial dimension (B, C, D=1, H, W)\n arr_torch = torch.from_numpy(arr).float().unsqueeze(0).unsqueeze(2)\n\n # Instantiate Rand3DElastic correctly\n transform = Rand3DElastic(\n prob=1.0,\n sigma_range=(5, 7), # controls smoothness of deformation\n magnitude_range=(50, 100), # deformation magnitude\n rotate_range=(0, 0, np.pi / 4), # rotation around the last axis only (Z-axis)\n scale_range=(0, 0.2, 0.2), # scale range (D,H,W)\n translate_range=(0, 50, 50), # translation (D,H,W)\n padding_mode='zeros',\n device=torch.device('cpu')\n )\n\n # Apply transform\n out_torch = transform(arr_torch)\n\n # Remove batch and dummy dimension: (B,C,D=1,H,W) -> (C,H,W)\n out_np = out_torch.squeeze(0).squeeze(1).cpu().numpy()\n\n # Channel-last conversion if needed\n if out_np.shape[0] == 3:\n out_np = np.transpose(out_np, (1, 2, 0))\n elif out_np.shape[0] == 1:\n out_np = out_np[0]\n\n # Clip values to valid image range and dtype\n out_np = np.clip(out_np, 0, 255).astype(img_obj.image.dtype)\n\n # Update the image object\n img_obj.set_image(out_np, method=\"random_elastic_deformation_monai\")\n return img_obj\n\n\ndef random_shear(img_obj, shear_range=(-0.2, 0.2)):\n \"\"\"\n Applies a random shear to the image.\n shear_range: tuple of (min_shear, max_shear) in radians.\n \"\"\"\n arr = img_obj.image\n\n # 1) Handle shape: (C,H,W) => (H,W,C)\n if arr.ndim == 3 and arr.shape[0] <= 4 and arr.shape[1] > 10:\n arr = np.transpose(arr, (1, 2, 0))\n\n # 2) Drop alpha if present\n if arr.ndim == 3 and arr.shape[-1] == 4:\n arr = arr[..., :3]\n\n # 3) Squeeze single-channel\n if arr.ndim == 3 and arr.shape[-1] == 1:\n arr = arr[..., 0]\n\n # Update the Image object with standardized array\n img_obj.image = arr\n\n # 4) Random shear factor\n shear_val = np.random.uniform(shear_range[0], shear_range[1])\n\n # Construct the AffineTransform\n transform = AffineTransform(shear=shear_val)\n\n try:\n # Warp the image\n sheared = warp(arr, transform, preserve_range=True, mode='constant', cval=0)\n sheared = np.clip(sheared, 0, 255).astype(arr.dtype)\n\n # 5) Update the image\n img_obj.set_image(sheared, method=f\"random_shear_{round(shear_val, 3)}\")\n except Exception as e:\n print(\"An error occurred while applying shear:\", e)\n\n return img_obj\n\ndef random_translate(img_obj, max_shift=(20, 20)):\n \"\"\"\n Translates the image randomly by up to +/- max_shift pixels (y_shift, x_shift).\n max_shift: tuple (max_shift_y, max_shift_x).\n \"\"\"\n arr = img_obj.image\n\n # 1) Handle shape: (C,H,W) => (H,W,C)\n if arr.ndim == 3 and arr.shape[0] <= 4 and arr.shape[1] > 10:\n arr = np.transpose(arr, (1, 2, 0))\n\n # 2) Drop alpha if present\n if arr.ndim == 3 and arr.shape[-1] == 4:\n arr = arr[..., :3]\n\n # 3) Squeeze single-channel\n if arr.ndim == 3 and arr.shape[-1] == 1:\n arr = arr[..., 0]\n\n img_obj.image = arr\n\n # 4) Sample random shifts in the range [-max_shift, +max_shift]\n shift_y = np.random.uniform(-max_shift[0], max_shift[0])\n shift_x = np.random.uniform(-max_shift[1], max_shift[1])\n\n # Construct the AffineTransform\n transform = AffineTransform(translation=(shift_x, shift_y))\n\n try:\n translated = warp(arr, transform, preserve_range=True, mode='constant', cval=0)\n translated = np.clip(translated, 0, 255).astype(arr.dtype)\n img_obj.set_image(translated, method=f\"random_translate_{(round(shift_x), round(shift_y))}\")\n except Exception as e:\n print(\"An error occurred while applying translation:\", e)\n\n return img_obj\n\ndef random_reflect(img_obj, prob=0.5, reflect_axes=('vertical', 'horizontal')):\n \"\"\"\n Randomly reflects (flips) the image across specified axes.\n reflect_axes can be a tuple/list of any combination of 'horizontal' / 'vertical'.\n For example, ('vertical',) only does vertical flips, ('horizontal',) only horizontal, \n ('vertical', 'horizontal') might do both with some probability logic, etc.\n prob is the probability that a reflection will happen at all.\n \"\"\"\n arr = img_obj.image\n\n # 1) Handle shape: (C,H,W) => (H,W,C)\n if arr.ndim == 3 and arr.shape[0] <= 4 and arr.shape[1] > 10:\n arr = np.transpose(arr, (1, 2, 0))\n\n # 2) Drop alpha if present\n if arr.ndim == 3 and arr.shape[-1] == 4:\n arr = arr[..., :3]\n\n # 3) Squeeze single-channel\n if arr.ndim == 3 and arr.shape[-1] == 1:\n arr = arr[..., 0]\n\n img_obj.image = arr\n\n if np.random.rand() < prob:\n # Decide which axis to flip\n # E.g., do vertical if 'vertical' in reflect_axes, do horizontal if 'horizontal' in reflect_axes\n flipped = arr.copy()\n description = []\n\n if 'horizontal' in reflect_axes and np.random.rand() < 0.5:\n flipped = np.flip(flipped, axis=1)\n description.append(\"H\")\n\n if 'vertical' in reflect_axes and np.random.rand() < 0.5:\n flipped = np.flip(flipped, axis=0)\n description.append(\"V\")\n\n img_obj.set_image(flipped, method=f\"random_reflect_{''.join(description)}\")\n\n return img_obj\n\ndef random_scale(img_obj, scale_range=(0.8, 1.2)):\n \"\"\"\n Scales (zooms) the image by a random factor in scale_range.\n e.g., (0.8, 1.2) means possible 20% contraction up to 20% expansion.\n \"\"\"\n arr = img_obj.image\n\n # 1) Handle shape: (C,H,W) => (H,W,C)\n if arr.ndim == 3 and arr.shape[0] <= 4 and arr.shape[1] > 10:\n arr = np.transpose(arr, (1, 2, 0))\n\n # 2) Drop alpha if present\n if arr.ndim == 3 and arr.shape[-1] == 4:\n arr = arr[..., :3]\n\n # 3) Squeeze single-channel\n if arr.ndim == 3 and arr.shape[-1] == 1:\n arr = arr[..., 0]\n\n img_obj.image = arr\n\n # 4) Random scale factor\n scale_factor = np.random.uniform(scale_range[0], scale_range[1])\n # If you want isotropic scaling, use the same factor for x and y:\n transform = AffineTransform(scale=(scale_factor, scale_factor))\n\n try:\n scaled = warp(arr, transform, preserve_range=True, mode='constant', cval=0)\n scaled = np.clip(scaled, 0, 255).astype(arr.dtype)\n img_obj.set_image(scaled, method=f\"random_scale_{round(scale_factor, 2)}\")\n except Exception as e:\n print(\"An error occurred while applying scaling:\", e)\n\n return img_obj\n \n\ndef spatial_pad(img_obj, new_size=(300, 300), mode=\"mean\"):\n \"\"\"\n Pad the image spatially to new_size = (target_height, target_width).\n If mode=\"mean\", pad with the mean pixel value.\n If mode=\"constant\", pad with 0 (or you can add another param).\n \"\"\"\n arr = img_obj.image\n\n # 1) Convert (C,H,W)->(H,W,C) if needed\n if arr.ndim == 3 and arr.shape[0] <= 4 and arr.shape[1] > 10:\n arr = np.transpose(arr, (1, 2, 0))\n\n # 2) Drop alpha channel if present\n if arr.ndim == 3 and arr.shape[-1] == 4:\n arr = arr[..., :3]\n\n # 3) Squeeze single-channel\n if arr.ndim == 3 and arr.shape[-1] == 1:\n arr = arr[..., 0]\n\n # Overwrite in the Image object\n img_obj.image = arr\n\n orig_h, orig_w = arr.shape[:2]\n pad_h, pad_w = new_size\n\n # If the image is already bigger than new_size, you can either skip\n # or do a center-crop, etc. For now, we'll just skip if bigger:\n if orig_h >= pad_h and orig_w >= pad_w:\n # no padding needed\n return img_obj\n\n # Determine pad values on each side so the image is (roughly) centered\n pad_top = (pad_h - orig_h) // 2 if pad_h > orig_h else 0\n pad_bottom = pad_h - orig_h - pad_top if pad_h > orig_h else 0\n pad_left = (pad_w - orig_w) // 2 if pad_w > orig_w else 0\n pad_right = pad_w - orig_w - pad_left if pad_w > orig_w else 0\n\n if mode == \"mean\":\n fill_value = arr.mean().astype(arr.dtype)\n else:\n # default to 0 if \"constant\" or unknown\n fill_value = 0\n\n # For grayscale: shape=(H,W), fill_value is scalar\n # For RGB: shape=(H,W,3), fill_value should be shape=(3,) or broadcastable\n if arr.ndim == 3:\n padded = np.full((pad_h, pad_w, arr.shape[2]), fill_value, dtype=arr.dtype)\n padded[pad_top:pad_top+orig_h, pad_left:pad_left+orig_w, :] = arr\n else:\n padded = np.full((pad_h, pad_w), fill_value, dtype=arr.dtype)\n padded[pad_top:pad_top+orig_h, pad_left:pad_left+orig_w] = arr\n\n img_obj.set_image(padded, method=f\"spatial_pad_{new_size}_{mode}\")\n return img_obj\n\n\ndef ben_graham_transform(img_obj, blur_ratio=0.1):\n \"\"\"\n Ben Graham's APTOS transform:\n 1) Optionally convert to HSV (depends on your preference).\n 2) Gaussian blur the image.\n 3) Merge original & blurred with addWeighted(...).\n \"\"\"\n arr = img_obj.image\n\n # Convert (C,H,W)->(H,W,C)\n if arr.ndim == 3 and arr.shape[0] <= 4 and arr.shape[1] > 10:\n arr = np.transpose(arr, (1, 2, 0))\n\n # If there's an alpha channel, drop it\n if arr.ndim == 3 and arr.shape[-1] == 4:\n arr = arr[..., :3]\n\n # If single-channel last dim => (H,W)\n if arr.ndim == 3 and arr.shape[-1] == 1:\n arr = arr[..., 0]\n\n # Overwrite\n img_obj.image = arr\n\n # Ensure we have color data for OpenCV. If grayscale, we'll skip HSV step\n # because HSV conversion requires 3 channels. \n is_color = (arr.ndim == 3 and arr.shape[-1] == 3)\n\n # Convert from RGB to BGR for OpenCV (if needed) \n # (OpenCV uses BGR ordering by default.)\n if is_color:\n arr_bgr = arr[..., ::-1].copy().astype(np.uint8) # RGB->BGR\n else:\n arr_bgr = arr.astype(np.uint8)\n\n # (Optionally) convert to HSV\n # arr_hsv = cv2.cvtColor(arr_bgr, cv2.COLOR_BGR2HSV) # uncomment if you want HSV\n \n # Let's just do the final \"Ben Graham\" line with BGR itself\n # Gaussian Blur\n radius = max(arr_bgr.shape[:2]) * blur_ratio # e.g. 0.1 of max dimension\n # OpenCV's GaussianBlur ksize must be odd & > 0. If radius=25.6 => ksize=25 or 27\n # We'll guess a kernel near radius*2+1 if it’s not huge\n k = int(radius*2 + 1)\n if k < 1:\n k = 1\n if k % 2 == 0:\n k += 1\n\n blurred = cv2.GaussianBlur(arr_bgr, (k, k), radius)\n\n # addWeighted => result = alpha*src1 + beta*src2 + gamma\n # from your snippet: 4*image + -4*blurred + 128\n # We must cast to float to avoid overflow\n out = cv2.addWeighted(arr_bgr, 4, blurred, -4, 128)\n\n # Convert back to RGB if color\n if is_color:\n out_rgb = out[..., ::-1]\n else:\n out_rgb = out\n\n out_rgb = out_rgb.astype(arr.dtype)\n\n # Put into image object\n img_obj.set_image(out_rgb, method=f\"ben_graham_transform\")\n\n return img_obj\n\ndef random_gaussian_noise(img_obj, mean=0, std=10, prob=0.5):\n \"\"\"\n Adds Gaussian noise (mean, std) to the image with probability = prob.\n \"\"\"\n arr = img_obj.image\n\n # Convert (C,H,W)->(H,W,C)\n if arr.ndim == 3 and arr.shape[0] <= 4 and arr.shape[1] > 10:\n arr = np.transpose(arr, (1, 2, 0))\n\n # Drop alpha\n if arr.ndim == 3 and arr.shape[-1] == 4:\n arr = arr[..., :3]\n\n # Squeeze single channel\n if arr.ndim == 3 and arr.shape[-1] == 1:\n arr = arr[..., 0]\n\n img_obj.image = arr\n\n if np.random.rand() < prob:\n # Generate noise\n noise = np.random.normal(mean, std, arr.shape)\n noisy = arr + noise\n noisy = np.clip(noisy, 0, 255).astype(arr.dtype)\n img_obj.set_image(noisy, method=f\"random_gaussian_noise_mean{mean}_std{std}\")\n\n return img_obj\n\n\ndef random_salt_pepper_noise(img_obj, amount=0.02, salt_vs_pepper=0.5, prob=0.5):\n \"\"\"\n Adds salt-and-pepper noise to the image.\n \n Parameters:\n - amount: fraction of pixels affected (e.g., 0.02 = 2% of pixels)\n - salt_vs_pepper: ratio of salt (white) pixels. 0.5 = equal salt and pepper.\n - prob: probability of applying this noise at all.\n \"\"\"\n arr = img_obj.image\n\n # 1) Handle (C,H,W) => (H,W,C)\n if arr.ndim == 3 and arr.shape[0] <= 4 and arr.shape[1] > 10:\n arr = np.transpose(arr, (1, 2, 0))\n\n # 2) Drop alpha channel if present\n if arr.ndim == 3 and arr.shape[-1] == 4:\n arr = arr[..., :3]\n\n # 3) Squeeze single-channel\n if arr.ndim == 3 and arr.shape[-1] == 1:\n arr = arr[..., 0]\n\n img_obj.image = arr\n\n # Decide if noise is applied based on prob\n if np.random.rand() < prob:\n noisy_img = np.copy(arr)\n total_pixels = arr.shape[0] * arr.shape[1]\n num_salt = np.ceil(amount * total_pixels * salt_vs_pepper).astype(int)\n num_pepper = np.ceil(amount * total_pixels * (1.0 - salt_vs_pepper)).astype(int)\n\n # Apply salt noise (white pixels)\n coords_salt = (\n np.random.randint(0, arr.shape[0], num_salt),\n np.random.randint(0, arr.shape[1], num_salt),\n )\n if arr.ndim == 2:\n noisy_img[coords_salt] = 255\n else:\n noisy_img[coords_salt[0], coords_salt[1], :] = 255\n\n # Apply pepper noise (black pixels)\n coords_pepper = (\n np.random.randint(0, arr.shape[0], num_pepper),\n np.random.randint(0, arr.shape[1], num_pepper),\n )\n if arr.ndim == 2:\n noisy_img[coords_pepper] = 0\n else:\n noisy_img[coords_pepper[0], coords_pepper[1], :] = 0\n\n img_obj.set_image(noisy_img, method=f\"random_salt_pepper_{amount}\")\n\n return img_obj","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-03-24T18:39:40.487874Z","iopub.execute_input":"2025-03-24T18:39:40.488524Z","iopub.status.idle":"2025-03-24T18:39:40.526348Z","shell.execute_reply.started":"2025-03-24T18:39:40.488498Z","shell.execute_reply":"2025-03-24T18:39:40.525621Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class Augmenter:\n def __init__(self, verbose=True):\n \"\"\"\n Initializes the Augmenter.\n \n Args:\n verbose (bool): If True, prints out details about operations.\n \"\"\"\n self.verbose = verbose\n self.last_id = 0\n # Store dataset (Collection) objects keyed by unique integer IDs\n self.datasets_dic = {}\n self.names_dic = {} # store corresponding dataset names\n\n def isunique(self, dataset_id):\n \"\"\"Check if a dataset_id is already used or not.\"\"\"\n return dataset_id not in self.datasets_dic\n\n def load_dataset_from_path(self, input_dir, dataset_name=None, dataset_id=None):\n \"\"\"\n Loads a dataset from a directory path into a Collection object\n and stores it in self.datasets_dic with a unique integer ID.\n\n Args:\n input_dir (str): File path to the dataset folder.\n dataset_name (str, optional): Name for the dataset. Defaults to folder name.\n dataset_id (int, optional): Custom ID for the dataset. If None,\n we attempt to use self.last_id (and increment as needed).\n \"\"\"\n if dataset_name is None:\n dataset_name = os.path.basename(input_dir.rstrip(\"/\"))\n\n # Auto-assign ID if not specified\n if dataset_id is None:\n while not self.isunique(self.last_id):\n self.last_id += 1\n dataset_id = self.last_id\n\n self.names_dic[dataset_id] = dataset_name\n\n dataset = Collection(\n directory_path=input_dir,\n name=dataset_name,\n verbose=self.verbose\n )\n self.datasets_dic[dataset_id] = dataset\n\n if self.verbose:\n print(f\"[Augmenter] Loaded dataset '{dataset_name}' \"\n f\"with ID={dataset_id} from '{input_dir}'.\")\n\n def display_first_images(self, n=1, ID=None):\n \"\"\"\n Convenience function: displays the first `n` images from each dataset in memory.\n \"\"\"\n for key, dataset in self.datasets_dic.items():\n if ID is None or key == ID:\n name = self.names_dic.get(key, f\"Unamed_{key}\")\n print(f\"--- Showing first {n} images for dataset \\\"{name}\\\" (ID: {key}) ---\")\n for x in range(n):\n image = dataset.images[x]\n image.show(title=f\"Dataset \\\"{name}\\\": Image #{x}\", axis=True)\n\n def create_new_dataset_id(self, proposed_name=\"augmented_dataset\"):\n \"\"\"\n Helper to create a new dataset ID (and store a name).\n Returns an integer dataset_id that is unique.\n \"\"\"\n while not self.isunique(self.last_id):\n self.last_id += 1\n new_id = self.last_id\n self.last_id += 1\n self.names_dic[new_id] = proposed_name\n return new_id\n\n # -----------------------------------------------------------------------\n # \"All-in-one\" method that:\n # 1) extracts raw images from an existing dataset,\n # 2) applies transformations behind the scenes,\n # 3) writes all results to disk in a new folder,\n # 4) optionally writes a CSV of filenames & label.\n #\n # The user only needs to know `dataset_id` and a few parameters.\n # They don't see any direct usage of 'extract_images' or how Collection\n # organizes images internally.\n # -----------------------------------------------------------------------\n def build_augmented_dataset_on_disk(self,\n dataset_id,\n output_dir,\n scale_factor=2.0,\n augmentations=None,\n label=None,\n to_df=True,\n debug=False,\n min_augmentations=1,\n max_augmentations=None,\n csv_name=\"dataset.csv\"):\n \"\"\"\n Higher-level convenience method:\n - Grabs raw images from a dataset (via extract_images).\n - Applies augmentations multiple times (scale_factor).\n - Writes each augmented image to a physical folder (output_dir).\n - If `to_df` is True, also writes a CSV with columns:\n [filename, label] if a label is provided\n or just [filename] if label=None.\n\n Args:\n dataset_id (int): Which dataset to augment/convert into a new disk folder.\n output_dir (str): Folder in which to store the augmented images.\n scale_factor (float): E.g. 2.0 => create 2 augmented images per original.\n augmentations (list, optional): List of augmentation functions \n (defaults to random_rotate + random_horizontal_flip).\n label (str, optional): If provided, put a 'label' column in the CSV.\n to_df (bool): If True, also generate a CSV of filenames (+ optional label).\n debug (bool): If each error should stop the program and display details\n min_augmentations (int): lower boundary for number of simultaneous augmentations per image\n max_augmentations (int or None): upper boundary\n csv_name (str): The CSV filename. Defaults to \"dataset.csv\".\n \"\"\"\n if max_augmentations is None:\n max_augmentations = len(augmentations)\n\n if max_augmentations > len(augmentations):\n max_augmentations = len(augmentations)\n \n if dataset_id not in self.datasets_dic:\n raise ValueError(f\"[Augmenter] Dataset with ID={dataset_id} not found.\")\n\n if not os.path.exists(output_dir):\n os.makedirs(output_dir, exist_ok=True)\n\n if augmentations is None:\n # Provide some default augmentations\n augmentations = [\n random_rotate, \n random_shear, \n random_translate, \n random_reflect, \n random_scale, \n #random_elastic_deformation, <-- WIP\n spatial_pad, \n ben_graham_transform, \n random_gaussian_noise,\n random_salt_pepper_noise,\n random_text_overlay\n ]\n\n collection_obj = self.datasets_dic[dataset_id]\n dataset_name = self.names_dic.get(dataset_id, f\"Dataset_{dataset_id}\")\n\n # 1) Extract the raw ND-arrays behind the scenes (the user doesn't see this)\n # If to_df=True, we get a DataFrame with each row being an array, plus a 'label' column if passed.\n # But in the standard implementation of Collection, extract_images(to_df=False)\n # simply returns a list of ND-arrays. We'll do that for simpler iteration.\n # (If you want label-based DataFrame usage, see the docstring in your Collection.)\n raw_images = collection_obj.extract_images(label=None, to_df=False)\n # raw_images is a list of NumPy arrays. \n # The dictionary keys are not returned, but we can keep a consistent naming approach.\n\n # 2) For each raw image, we generate `scale_factor` augmentations\n # and write them to disk. We also keep track of filenames for CSV if needed.\n n_aug_per_img = int(scale_factor)\n csv_rows = [] # to store (filename, label)\n\n for idx, np_img in enumerate(tqdm(raw_images)):\n base_name = f\"{dataset_name}_img{idx}\"\n \n for i in range(n_aug_per_img):\n try:\n # Make a fresh Image (from fast_skimage) for each iteration\n new_img = Image(np.copy(np_img), chatty_mode=False)\n new_img.name = base_name\n \n # Apply random subset of augmentations\n num_transforms_to_apply = random.randint(min_augmentations, max_augmentations)\n random_transforms = random.sample(augmentations, num_transforms_to_apply)\n for aug_func in random_transforms:\n new_img = aug_func(new_img)\n \n # Instead of new_img.save_image_data(...), \n # just call imsave on new_img.image:\n out_filename = f\"{new_img.name}_aug_{idx}_{i}.jpg\"\n out_path = os.path.join(output_dir, out_filename)\n \n # Write as a standard PNG:\n imsave(out_path, new_img.image)\n \n # Optionally record in CSV:\n if to_df:\n if label is not None:\n csv_rows.append((out_filename, label))\n else:\n csv_rows.append((out_filename,))\n except Exception as e:\n if debug:\n raise(e)\n print(\"Could not augment image\", i, \"because of an error:\", e)\n \n # 3) Optionally write a CSV summarizing all output filenames\n if to_df:\n if label is not None:\n df = pd.DataFrame(csv_rows, columns=[\"filename\", \"label\"])\n else:\n df = pd.DataFrame(csv_rows, columns=[\"filename\"])\n\n csv_out_path = os.path.join(output_dir, csv_name)\n df.to_csv(csv_out_path, index=False)\n if self.verbose:\n print(f\"[Augmenter] CSV with {len(csv_rows)} entries saved to '{csv_out_path}'.\")\n\n if self.verbose:\n print(f\"[Augmenter] Built augmented dataset from dataset_id={dataset_id}\")\n print(f\" => {output_dir} (scale_factor={scale_factor}, #images={len(raw_images)*n_aug_per_img}).\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-03-24T18:39:40.528036Z","iopub.execute_input":"2025-03-24T18:39:40.528355Z","iopub.status.idle":"2025-03-24T18:39:40.552240Z","shell.execute_reply.started":"2025-03-24T18:39:40.528322Z","shell.execute_reply":"2025-03-24T18:39:40.551590Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"\n
\n
5. Apply data augmentation to benchmark datasets
\n
\n\n[Back to table of contents](#TOC)\n\nMNIST is loaded in a csv file, so first we need to extract it:","metadata":{}},{"cell_type":"code","source":"#| export\n\ndata = pd.read_csv(\"/kaggle/input/digit-recognizer/train.csv\")\ndata.describe()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-03-24T18:39:40.553545Z","iopub.execute_input":"2025-03-24T18:39:40.553817Z","iopub.status.idle":"2025-03-24T18:39:45.001790Z","shell.execute_reply.started":"2025-03-24T18:39:40.553784Z","shell.execute_reply":"2025-03-24T18:39:45.001129Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#| export\n\n# This function was initially built to load MNIST, but you can use it on other datasets\n\ndef load_images_from_csv(\n csv_path,\n label_column=\"label\",\n output_data_dir=\"Dataset\",\n limit=None,\n normalize=False\n):\n \"\"\"\n Loads a dataset from a CSV with one 'label' (or custom label_column) \n and multiple pixel columns (e.g., pixel0..pixel783), then writes each \n row to an image file in /