Data_Pre_Processing.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import cv2\n",
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "import os\n",
    "from scipy import signal\n",
    "from scipy.fft import fftshift\n",
    "from matplotlib.pyplot import figure, show\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "def getaudiodict(directory):\n",
    "    \n",
    "    audio_directory = directory + 'Recording/'\n",
    "    audio_dictionary = {}\n",
    "\n",
    "    for filename in os.listdir(audio_directory):\n",
    "        # Get Filename Prefix\n",
    "        filename_prefix = filename.split('.')[0]\n",
    "        # Import the audio sample\n",
    "        recorded_sample = np.load(os.path.join(audio_directory, filename))\n",
    "\n",
    "        # Create a channels dictionary\n",
    "        channels_data = {}\n",
    "        for x in range(6):\n",
    "            channels_data[\"channel_{0}\".format(x+1)] = recorded_sample.reshape([-1,6])[:,x]\n",
    "        if np.max(channels_data['channel_1']) < 1200:\n",
    "            continue\n",
    "        else:\n",
    "            # Get Stacked Spectrogram \n",
    "            fs = 48000  # Sampling Rate \n",
    "            duration = len(channels_data['channel_1'])/fs\n",
    "            original_features = []\n",
    "\n",
    "            for channel_no in channels_data:         \n",
    "                #Slicing the channel sample \n",
    "                max_index = np.where(channels_data[channel_no] == np.max(channels_data[channel_no]))[0][0]\n",
    "                start = max_index + int(len(channels_data[channel_no])*0.0015 / duration)\n",
    "                end = start + int(len(channels_data[channel_no])*0.03 / duration)\n",
    "                sliced_data = channels_data[channel_no][start:end]\n",
    "\n",
    "                # Create Spectrogram\n",
    "                f, t, Sxx = signal.spectrogram(sliced_data, fs, nperseg=256)\n",
    "\n",
    "                # Stack Spectrograms\n",
    "                original_features.append(Sxx)\n",
    "            \n",
    "            condition_list = []\n",
    "            for i in original_features:\n",
    "                condition_list.append(i.shape == (129, 6))\n",
    "            if all(condition_list):\n",
    "                stacked_sample = np.stack(original_features, axis=0)\n",
    "                audio_dictionary[filename_prefix] = stacked_sample\n",
    "            else:\n",
    "                continue\n",
    "\n",
    "    return audio_dictionary\n",
    "    \n",
    "def getlabeldict(directory):\n",
    "\n",
    "    images_directory = directory + 'Images/'\n",
    "    label_dictionary = {}\n",
    "    error_files = []\n",
    "    \n",
    "    for filename in os.listdir(images_directory):\n",
    "        filename_prefix = filename.split('.')[0]\n",
    "\n",
    "        img = cv2.imread(os.path.join(images_directory, filename))\n",
    "        img_cropped = img[162:304, 231:383]\n",
    "        gray = cv2.cvtColor(img_cropped,cv2.COLOR_BGR2GRAY)\n",
    "        gray = np.float32(gray)\n",
    "\n",
    "        corners = cv2.goodFeaturesToTrack(gray,1,0.01,15)\n",
    "#         corners = cv2.goodFeaturesToTrack(gray,2,0.01,5)    # Corner detection parameters for 2 objects \n",
    "        corners = np.int0(corners)\n",
    "        corners_tuple = []\n",
    "\n",
    "        for i in corners:\n",
    "            x, y = i[0][0], i[0][1]\n",
    "            corners_tuple.append((x,y))\n",
    "\n",
    "        if len(corners_tuple) == 1:\n",
    "            x_coordinate = round((corners_tuple[0][0] / img_cropped.shape[1]), 3)\n",
    "            y_coordinate = round((corners_tuple[0][1] / img_cropped.shape[0]), 3)\n",
    "            coordiantes = [x_coordinate, y_coordinate]\n",
    "            \n",
    "#           Uncomment the following for two object detection\n",
    "#             x_coordinate_1 = round((corners_tuple[0][0] / img_cropped.shape[1]), 3)\n",
    "#             y_coordinate_1 = round((corners_tuple[0][1] / img_cropped.shape[0]), 3)\n",
    "#             x_coordinate_2 = round((corners_tuple[1][0] / img_cropped.shape[1]), 3)\n",
    "#             y_coordinate_2 = round((corners_tuple[1][1] / img_cropped.shape[0]), 3)\n",
    "#             coordiantes = [x_coordinate_1, y_coordinate_1, x_coordinate_2, y_coordinate_2]\n",
    "\n",
    "            label_dictionary[filename_prefix] = coordiantes\n",
    "        else:\n",
    "            error_files.append(filename_prefix)\n",
    "        \n",
    "    return label_dictionary\n",
    "\n",
    "def getXandY(audio_dict, label_dict):\n",
    "    x = []\n",
    "    y = []\n",
    "    for filename in audio_dict:\n",
    "        if filename in label_dict:\n",
    "            x.append(audio_dict[filename])\n",
    "            y.append(label_dict[filename])\n",
    "        else:\n",
    "            continue\n",
    "    X = np.stack(x, axis=0)\n",
    "    Y = np.stack(y, axis=0)\n",
    "    return X, Y"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "time taken for creating X and Y matrices is  2.8427886525789896 minutes\n"
     ]
    }
   ],
   "source": [
    "# Getting X and Y for Model v1. FFT size = 128 and Spectrogram size = (65, 12)\n",
    "\n",
    "start = time.time()\n",
    "directory_name = r'Enter your directory name'\n",
    "audio_d = getaudiodict(directory_name)\n",
    "label_d = getlabeldict(directory_name)\n",
    "X, Y = getXandY(audio_d, label_d)\n",
    "assert(X.shape[0] == Y.shape[0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(1497, 6, 129, 6) (1497, 2)\n"
     ]
    }
   ],
   "source": [
    "# Save created X and Y to respective npy files\n",
    "\n",
    "np.save('Enter X Matrix Name', X)\n",
    "np.save('Enter Y Matrix Name', Y)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "thesis",
   "language": "python",
   "name": "thesis"
  },
  "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.7.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}