image_preprocessing.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "import skimage\n",
    "from skimage import io\n",
    "import matplotlib.pyplot as plt\n",
    "import json\n",
    "import cv2\n",
    "import numpy as np\n",
    "import os\n",
    "from tqdm import tqdm\n",
    "import matplotlib.patches as patches\n",
    "import matplotlib"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "outputs": [],
   "source": [
    "data_dir = os.path.join('data_WiSAR','data')\n",
    "data_dir = os.path.join('data')"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [],
   "source": [
    "def preprocess_image(img, mask, homography):\n",
    "    '''\n",
    "    cuts out provided mask for timestamp\n",
    "    warps image to the perspective of the center camera with provided homography matrix\n",
    "    '''\n",
    "    img = img[..., :3].astype(np.float32)/255\n",
    "    img[mask==0] = np.nan\n",
    "    processed_img = img.copy()\n",
    "    for dim in range(3):\n",
    "        processed_img[:,:,dim] = cv2.warpPerspective(img[:,:,dim],homography,img.shape[:2],borderMode = cv2.BORDER_CONSTANT,borderValue = np.nan) \n",
    "    return processed_img"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "def integrate_images(img_list):\n",
    "    '''\n",
    "    averages all images in img_list\n",
    "    removes pixels of the average image that are not covered by all images\n",
    "    '''\n",
    "    integrated_img = np.nanmean(np.array(img_list), axis = 0)\n",
    "    integrated_img_nan = integrated_img.copy()\n",
    "    for dim in range(3):\n",
    "        integrated_img_nan[:,:,dim] = np.where(np.isnan(img_list).any(axis = 0).all(axis = 2), np.nan, integrated_img[:,:,dim])\n",
    "    return integrated_img_nan"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [],
   "source": [
    "def crop_image(img):\n",
    "    '''\n",
    "    crops images to a hardcoded size (315x700), so that no nan-values occur in any of the images\n",
    "    '''\n",
    "    cropped_img = img[370:685,140:840,:]\n",
    "    return cropped_img"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [],
   "source": [
    "def label_image(img,label,shift = False,max_pixel = 1,dim=3):\n",
    "    '''\n",
    "    labels image with provided labels as red rectangles\n",
    "    shift: set True for cropped images and False for images in original size\n",
    "    '''\n",
    "    if shift:\n",
    "        x_shift, y_shift = 140, 370\n",
    "    else:\n",
    "        x_shift, y_shift = 0, 0\n",
    "    for lab in label:\n",
    "        x = lab[0]-x_shift\n",
    "        y = lab[1]-y_shift\n",
    "        w = lab[2]\n",
    "        h = lab[3]\n",
    "        if dim == 3:\n",
    "            for i in range(x, x+w):\n",
    "                img[y,i,:] = [max_pixel]*3\n",
    "                img[y+h,i,:] = [max_pixel]*3\n",
    "            for j in range(y, y+h):\n",
    "                img[j,x,:] = [max_pixel]*3\n",
    "                img[j,x+w,:] = [max_pixel]*3\n",
    "        elif dim == 2:\n",
    "            for i in range(x, x+w):\n",
    "                img[y,i] = max_pixel\n",
    "                img[y+h,i] = max_pixel\n",
    "            for j in range(y, y+h):\n",
    "                img[j,x] = max_pixel\n",
    "                img[j,x+w] = max_pixel\n",
    "    return img"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [],
   "source": [
    "def apply_brightness_contrast(input_img, brightness = 0, contrast = 0):\n",
    "    \n",
    "    if brightness != 0:\n",
    "        if brightness > 0:\n",
    "            shadow = brightness\n",
    "            highlight = 255\n",
    "        else:\n",
    "            shadow = 0\n",
    "            highlight = 255 + brightness\n",
    "        alpha_b = (highlight - shadow)/255\n",
    "        gamma_b = shadow\n",
    "        \n",
    "        buf = cv2.addWeighted(input_img, alpha_b, input_img, 0, gamma_b)\n",
    "    else:\n",
    "        buf = input_img.copy()\n",
    "    \n",
    "    if contrast != 0:\n",
    "        f = 131*(contrast + 127)/(127*(131-contrast))\n",
    "        alpha_c = f\n",
    "        gamma_c = 127*(1-f)\n",
    "        \n",
    "        buf = cv2.addWeighted(buf, alpha_c, buf, 0, gamma_c)\n",
    "\n",
    "    return buf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [],
   "source": [
    "def process_data(dataset, save = False, plot = True, label = False, only_cropped = False):\n",
    "    '''\n",
    "    creates integrated and cropped images as well as images showing the variance between 7 timesteps \n",
    "    dataset: 'train', 'test' or 'validation'\n",
    "    save: set to True if images should be saved in subfolders of 'data_WiSAR/data'\n",
    "    plot: set to True if images should be plotted in the jupyter notebook\n",
    "    label: set to True if images should be labelled given that labels are provided - labelled images get\n",
    "        saved in a separate folder\n",
    "    only_cropped: Save only the cropped images\n",
    "    '''\n",
    "\n",
    "    mask = skimage.io.imread(os.path.join(data_dir,'mask.png'))\n",
    "    if label:\n",
    "        try:\n",
    "            with open(os.path.join(data_dir,dataset,'labels.json')) as f:\n",
    "                labels = json.load(f)\n",
    "        except:\n",
    "            label = False\n",
    "    labelled_flag = ''\n",
    "    if label:\n",
    "        labelled_flag = '_labelled'\n",
    "    if save:\n",
    "        if not os.path.isdir(os.path.join(data_dir,'processed_images_'+dataset+labelled_flag)):\n",
    "            os.makedirs(os.path.join(data_dir,'processed_images_'+dataset+labelled_flag))\n",
    "    for folder in os.listdir(os.path.join(data_dir, dataset)):\n",
    "        if folder.startswith(dataset) or folder.startswith('valid'):\n",
    "            if save:\n",
    "                save_folder = os.path.join(data_dir,'processed_images_'+dataset+labelled_flag,folder)\n",
    "                if not os.path.isdir(save_folder):\n",
    "                    os.makedirs(save_folder)\n",
    "            with open(os.path.join(data_dir,dataset,folder,'homographies.json')) as f:\n",
    "                homographies = json.load(f)\n",
    "            if label:\n",
    "                label_coords = labels[folder]\n",
    "            all_processed_images = []\n",
    "            cropped_images = []\n",
    "            for timepoint in [0,1,2,3,4,5,6]:\n",
    "                processed_images = []\n",
    "                for camera in ['B01', 'B02', 'B03', 'B04', 'B05', 'G01', 'G02', 'G03', 'G04', 'G05']:\n",
    "                    image = skimage.io.imread(os.path.join(data_dir,dataset,folder,str(timepoint)+'-'+camera+'.png'))\n",
    "                    homography = np.array(homographies[str(timepoint) + '-' + camera])\n",
    "                    processed_image = preprocess_image(image, mask, homography)\n",
    "                    processed_images.append(processed_image)\n",
    "                    all_processed_images.append(processed_image)\n",
    "                integrated_image = integrate_images(processed_images)\n",
    "                cropped_image = crop_image(integrated_image)\n",
    "                cropped_images.append(cropped_image)\n",
    "                if label:\n",
    "                    integrated_image = label_image(integrated_image, label = label_coords)\n",
    "                    cropped_image = label_image(cropped_image, label = label_coords, shift = True)\n",
    "\n",
    "                if save and not only_cropped:\n",
    "                    matplotlib.image.imsave(os.path.join(save_folder,'integrated_image_'+str(timepoint)+'.png'), integrated_image)\n",
    "                    matplotlib.image.imsave(os.path.join(save_folder,'cropped_image_'+str(timepoint)+'.png'), cropped_image)\n",
    "\n",
    "                if plot:\n",
    "                    plt.figure()\n",
    "                    plt.imshow(integrated_image)\n",
    "                    plt.figure()\n",
    "                    plt.imshow(cropped_image)            \n",
    "\n",
    "            integrated_image_all = integrate_images(all_processed_images)\n",
    "            cropped_image_all = crop_image(integrated_image_all)\n",
    "            var_image = np.nanvar(cropped_images, axis = 0)/np.max(np.nanvar(cropped_images, axis = 0))\n",
    "                      \n",
    "            if label:\n",
    "                integrated_image_all = label_image(integrated_image_all, label = label_coords)\n",
    "                cropped_image_all = label_image(cropped_image_all, label = label_coords, shift = True)\n",
    "                var_image = label_image(var_image, label = label_coords, shift = True)\n",
    "\n",
    "            if save and not only_cropped:\n",
    "                matplotlib.image.imsave(os.path.join(save_folder,'integrated_image_all.png'), integrated_image_all)\n",
    "                matplotlib.image.imsave(os.path.join(save_folder,'cropped_image_all.png'), cropped_image_all)\n",
    "                matplotlib.image.imsave(os.path.join(save_folder,'image_var.png'), var_image)\n",
    "            if save and only_cropped:\n",
    "                matplotlib.image.imsave(os.path.join(save_folder,'cropped_image_all.png'), cropped_image_all)\n",
    "\n",
    "            if plot:\n",
    "                plt.figure()\n",
    "                plt.imshow(integrated_image_all)\n",
    "                plt.figure()\n",
    "                plt.imshow(cropped_image_all)   \n",
    "                plt.figure()\n",
    "                plt.imshow(var_image)\n",
    "                plt.figure()\n",
    "                plt.imshow(var_threshold_image) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_cropped_dataset(dataset):\n",
    "    '''\n",
    "    saves all cropped images from process_data() in one folder per dataset\n",
    "    dataset: 'train', 'test', 'validation', 'validation_labelled'\n",
    "    requires process_data(dataset, save = False, plot = True, label = False) to be run first\n",
    "    '''\n",
    "    save_path = os.path.join(data_dir,'cropped_dataset',dataset)\n",
    "    if not os.path.isdir(save_path):\n",
    "        os.makedirs(save_path)\n",
    "    for folder in os.listdir(os.path.join(data_dir, 'processed_images_' + dataset)):\n",
    "        for timepoint in [0,1,2,3,4,5,6]:\n",
    "            cropped_image = skimage.io.imread(os.path.join(data_dir,'processed_images_' + dataset,folder,'cropped_image_'+str(timepoint)+'.png'))\n",
    "            cropped_image = cropped_image[..., :3].astype(np.float32)/255\n",
    "            matplotlib.image.imsave(os.path.join(save_path,folder+'_'+'cropped_'+str(timepoint)+'.png'), cropped_image)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_variance_dataset(dataset):\n",
    "    '''\n",
    "    saves all variance images from process_data() in one folder per dataset\n",
    "    dataset: 'train', 'test', 'validation', 'validation_labelled'\n",
    "    requires process_data(dataset, save = False, plot = True, label = False) to be run first\n",
    "    '''\n",
    "    save_path = os.path.join(data_dir,'variance_dataset',dataset)\n",
    "    if not os.path.isdir(save_path):\n",
    "        os.makedirs(save_path)\n",
    "    for folder in os.listdir(os.path.join(data_dir, 'processed_images_' + dataset)):\n",
    "        var_image = skimage.io.imread(os.path.join(data_dir,'processed_images_' + dataset,folder,'image_var.png'))\n",
    "        var_image = var_image[..., :3].astype(np.float32)/255\n",
    "        matplotlib.image.imsave(os.path.join(save_path,folder+'_variance.png'), var_image)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_variance_channel_dataset(dataset, colour):\n",
    "    '''\n",
    "    saves the selected colour channel of the variance images from create_variance_dataset() as greyscale image \n",
    "    as well as the subtraction of the green channel from the selected channel in two different folders per dataset\n",
    "    dataset: 'train', 'test', 'validation', 'validation_labelled'\n",
    "    colour: 'blue', 'red'\n",
    "    requires create_variance_dataset(dataset, colour) to be run first\n",
    "    '''\n",
    "    save_path_colour = os.path.join(data_dir,'variance_dataset_'+colour,dataset)\n",
    "    if not os.path.isdir(save_path_colour):\n",
    "        os.makedirs(save_path_colour)\n",
    "    save_path_colour_green = os.path.join(data_dir,'variance_dataset_'+colour+'-green',dataset)\n",
    "    if not os.path.isdir(save_path_colour_green):\n",
    "        os.makedirs(save_path_colour_green)\n",
    "    if dataset == 'validation_labelled':\n",
    "        with open(os.path.join(data_dir,'validation','labels.json')) as f:\n",
    "            labels = json.load(f)\n",
    "    if colour == 'red':\n",
    "        channel = 0\n",
    "    elif colour == 'blue':\n",
    "        channel = 2\n",
    "    for folder in os.listdir(os.path.join(data_dir, 'processed_images_' + dataset)):\n",
    "                \n",
    "        var_image = skimage.io.imread(os.path.join(data_dir,'variance_dataset',dataset,folder+'_variance.png'))\n",
    "        var_image = var_image[..., :3].astype(np.float32)/255\n",
    "        channel_image = var_image[...,channel]\n",
    "        matplotlib.image.imsave(os.path.join(save_path_colour,folder+'_variance_'+colour+'.png'), channel_image, cmap='gray')\n",
    "\n",
    "        green_image = var_image[...,1]\n",
    "        green_sub_image = channel_image - green_image\n",
    "        green_sub_image = np.where(green_sub_image<0, 0, green_sub_image)\n",
    "        green_sub_image /= np.max(green_sub_image)\n",
    "        if dataset == 'validation_labelled':\n",
    "            label_coords = labels[folder]\n",
    "            green_sub_image = label_image(green_sub_image, label = label_coords, shift = True, dim = 2)\n",
    "\n",
    "        matplotlib.image.imsave(os.path.join(save_path_colour_green,folder+'_variance_'+colour+'-green.png'), green_sub_image, cmap='gray')\n",
    "        "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_variance_green_dataset(dataset):\n",
    "    '''\n",
    "    saves the green channel of the variance images from create_variance_dataset() as greyscale images \n",
    "    not relevant for analysis but included for the pupose of completeness (used in powerpoint presentation)\n",
    "    dataset: 'train', 'test', 'validation', 'validation_labelled'\n",
    "    requires create_variance_dataset(dataset, colour) to be run first\n",
    "    '''\n",
    "    save_path_colour = os.path.join(data_dir,'variance_dataset_green',dataset)\n",
    "    if not os.path.isdir(save_path_colour):\n",
    "        os.makedirs(save_path_colour)\n",
    "    if dataset == 'validation_labelled':\n",
    "        with open(os.path.join(data_dir,'validation','labels.json')) as f:\n",
    "            labels = json.load(f)\n",
    "\n",
    "    for folder in os.listdir(os.path.join(data_dir, 'processed_images_' + dataset)):\n",
    "                \n",
    "        var_image = skimage.io.imread(os.path.join(data_dir,'variance_dataset',dataset,folder+'_variance.png'))\n",
    "        var_image = var_image[..., :3].astype(np.float32)/255\n",
    "        channel_image = var_image[...,1]\n",
    "        matplotlib.image.imsave(os.path.join(save_path_colour,folder+'_variance_green.png'), channel_image, cmap='gray')\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_contrast_brightness_dataset(dataset,colour,contrast=96,brightness=32):\n",
    "    '''\n",
    "    adjusts contrast and brightness on the green subtracted create_variance_channel_dataset()\n",
    "    and saves them in a folder per dataset \n",
    "    dataset: 'train', 'test', 'validation', 'validation_labelled'\n",
    "    colour: 'blue', 'red'\n",
    "    contrast: int between -128 and 128, default: 96\n",
    "    brightness: int between -128 and 128, default: 32\n",
    "    requires create_variance_channel_dataset(dataset, colour) to be run first\n",
    "    '''\n",
    "    save_path = os.path.join(data_dir,'variance_dataset_'+colour+'-green_adjusted',dataset)\n",
    "    if dataset == 'validation_labelled':\n",
    "        with open(os.path.join(data_dir,'validation','labels.json')) as f:\n",
    "            labels = json.load(f)\n",
    "    if not os.path.isdir(save_path):\n",
    "        os.makedirs(save_path)\n",
    "    for folder in os.listdir(os.path.join(data_dir, 'processed_images_' + dataset)):\n",
    "        if dataset == 'validation_labelled':\n",
    "            label_coords = labels[folder]\n",
    "\n",
    "        green_sub_image = skimage.io.imread(os.path.join(data_dir,'variance_dataset_'+colour+'-green',dataset,folder+'_variance_'+colour+'-green.png'))\n",
    "        green_sub_image = green_sub_image[..., :3]\n",
    "\n",
    "        adjusted_image = apply_brightness_contrast(green_sub_image, brightness, contrast)\n",
    "        if dataset == 'validation_labelled':\n",
    "            adjusted_image = label_image(adjusted_image, label = label_coords, shift = True, max_pixel = 255)\n",
    "        matplotlib.image.imsave(os.path.join(save_path,folder+'_variance_'+colour+'-green_c'+str(contrast)+'_b'+str(brightness)+'.png'), adjusted_image, cmap='gray')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/var/folders/g8/7dsc7c7d5757b7m6xbqnwy680000gn/T/ipykernel_52366/1819771478.py:6: RuntimeWarning: Mean of empty slice\n",
      "  integrated_img = np.nanmean(np.array(img_list), axis = 0)\n"
     ]
    }
   ],
   "source": [
    "# run all\n",
    "\n",
    "process_data('train', save = True, plot = False, label = False, only_cropped=False)\n",
    "process_data('test', save = True, plot = False, label = False, only_cropped=False)\n",
    "process_data('validation', save = True, plot = False, label = False, only_cropped=False)\n",
    "process_data('validation', save = True, plot = False, label = True, only_cropped=False)\n",
    "\n",
    "for dataset in ['train', 'test', 'validation', 'validation_labelled']:\n",
    "    create_cropped_dataset(dataset)\n",
    "    create_variance_dataset(dataset)\n",
    "#     create_variance_green_dataset(dataset)\n",
    "    for colour in ['blue', 'red']:\n",
    "        create_variance_channel_dataset(dataset, colour)\n",
    "        create_contrast_brightness_dataset(dataset, colour)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/var/folders/g8/7dsc7c7d5757b7m6xbqnwy680000gn/T/ipykernel_52366/1819771478.py:6: RuntimeWarning: Mean of empty slice\n",
      "  integrated_img = np.nanmean(np.array(img_list), axis = 0)\n"
     ]
    }
   ],
   "source": [
    "process_data('train', save = True, plot = False, label = False, only_cropped=True)\n",
    "process_data('test', save = True, plot = False, label = False, only_cropped=True)\n",
    "process_data('validation', save = True, plot = False, label = False, only_cropped=True)"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}