Radio_Modulation_Recognition_Networks.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Radio_Modulation_Recognition_Networks.ipynb",
      "provenance": [],
      "collapsed_sections": [
        "njyrqD4TFG4K",
        "vs1M0dXbEkvf",
        "sA1QnRJ5aNIU",
        "S3tzTIiCyHd8",
        "ZfPb1pzNFzJZ",
        "9H6_aAZ9tTiB"
      ],
      "toc_visible": true,
      "authorship_tag": "ABX9TyNcsg7ev79m2QRGatYlokaT",
      "include_colab_link": true
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/KristynaPijackova/Radio-Modulation-Recognition-Networks/blob/main/Radio_Modulation_Recognition_Networks.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "zad6iuE6ULe-"
      },
      "source": [
        "# Radio Modulation Recognition Networks\n",
        "\n",
        "---\n",
        "\n",
        "**Author: Kristyna Pijackova**\n",
        "\n",
        "---\n",
        "\n",
        "This notebook contains code for my [bachelor thesis](https://www.vutbr.cz/studenti/zav-prace/detail/133594) in the academic year 2020/2021. \n",
        "\n",
        "---\n",
        "\n",
        "**The code structure is following:**\n",
        "\n",
        "\n",
        "*   **Imports** - Import needed libraries\n",
        "*   **Defined Functions** - Functions defined for an easier manipulation with the data later on\n",
        "*   **Accessing the datasets** - you may skip this part and download the datasets elsewhere if you please\n",
        "*   **Loading Data** - Load the data and divide them into training, validation and test sets\n",
        "*   **Deep Learning Part** -Contains the architectures, which are prepared to be trained and evaluated\n",
        "*   **Load Trained Model** - Optionaly you can download the CGDNN model and see how it does on the corresponding dataset\n",
        "*   **Layer Visualization** - A part of code which was written to visualize the activation maps of the convolutional and recurrent layers\n",
        "*   **Plotting** - You can plot the confusion matrices in this part \n",
        "\n",
        "---\n",
        "\n",
        "**Quick guide to running the document:**\n",
        "\n",
        "Open  [Google Colab](https://colab.research.google.com/notebooks/intro.ipynb#recent=true) and go to 'GitHub' bookmark. Insert the link to the Github repository. This should open the code for you and allow you to run and adjust it.\n",
        "\n",
        "*   Use `up` and `down` keys to move in the notebook\n",
        "*   Use `ctrl+enter` to run cell or choose 'Run All' in Runtime to run the whole document at once \n",
        "*   If you change something in specific cell, it's enough to re-run just the cell to save the changes\n",
        "*   Hide/show sections of the code with the arrows at side, which are next to some cell code\n",
        "* In the top left part yoz can click on the Content icon, which will allow you to navigate easier through this notebook\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "3RmHw6yPDVu5"
      },
      "source": [
        "# Imports"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "EVuoveNXEo-Z"
      },
      "source": [
        "Import needed libraries"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "l84_k5S5Qls-"
      },
      "source": [
        "from scipy.io import loadmat\n",
        "from pandas import factorize\n",
        "import pickle\n",
        "import numpy as np\n",
        "import random\n",
        "from scipy import signal\n",
        "\n",
        "from matplotlib import pyplot as plt\n",
        "from sklearn.metrics import confusion_matrix\n",
        "import seaborn as sns\n",
        "\n",
        "from tensorflow.keras.utils import to_categorical\n",
        "import tensorflow as tf\n",
        "from tensorflow import keras\n",
        "from tensorflow.keras import backend as K\n",
        "from tensorflow.keras import layers\n",
        "from tensorflow.keras.utils import plot_model"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "mgQ2Bed7EifS"
      },
      "source": [
        "Mount to Google Drive (optional)"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "RHY8ixKOQw12"
      },
      "source": [
        "# Mounting your Google Drive\n",
        "\n",
        "# from google.colab import drive\n",
        "# drive.mount('/content/gdrive', force_remount=True)\n",
        "# root_dir = \"/content/gdrive/My Drive/\""
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "njyrqD4TFG4K"
      },
      "source": [
        "# Defined functions for easier work with data"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ZefCAE3tEh_v"
      },
      "source": [
        "## Functions to load datasets"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "KXqg2ddj_U8i"
      },
      "source": [
        "# VUT Dataset\n",
        "def load_VUT_dataset(dataset_location):\n",
        "    \"\"\"\n",
        "    Load dataset and extract needed data\n",
        "\n",
        "    Input: \n",
        "        dataset_location: specify where the file is stored and its name\n",
        "\n",
        "    Output:\n",
        "        SNR: list of the SNR range in dataset [-20 to 18]\n",
        "        X: array of the measured I/Q data [num_of_samples, 128, 2]\n",
        "        modulations: list of the modulations in this dataset\n",
        "        one_hot: one_hot encoded data - the other maps the order of the mods\n",
        "        lbl_SNR: list of each snr (for plotting)\n",
        "    \"\"\"\n",
        "\n",
        "    # Load the dataset stored as .mat with loadmat fuction from scipy.io\n",
        "\n",
        "    # from scipy.io import loadmat\n",
        "    dataset = loadmat(dataset_location)\n",
        "\n",
        "    # Point to wanted data\n",
        "\n",
        "    SNR = dataset['SNR']\n",
        "    X = dataset['X']    \n",
        "    mods = dataset['mods']\n",
        "    one_hot = dataset['one_hot']    \n",
        "\n",
        "    # Transpose the structure of X from [:,2,128] to [:,128,2]\n",
        "\n",
        "    X = np.transpose(X[:,:,:],(0,2,1))\n",
        "\n",
        "    # Change the type and structure of output SNR and mods to lists\n",
        "\n",
        "    SNRs = []\n",
        "    SNR = np.reshape(SNR,-1)\n",
        "\n",
        "    for i in range(SNR.shape[0]):\n",
        "        snr = SNR[:][i].tolist()\n",
        "        SNRs.append(snr)\n",
        "\n",
        "    modulations = []\n",
        "    mods = np.reshape(mods,-1)\n",
        "\n",
        "    for i in range(mods.shape[0]):\n",
        "        mod = mods[i][0].tolist()\n",
        "        modulations.append(mod)\n",
        "\n",
        "    # Assign SNR value to each vector\n",
        "    repeat_n = X.shape[0]/len(mods)/len(SNR)\n",
        "    repeat_n_mod = len(mods)    \n",
        "    lbl_SNR = np.tile(np.repeat(SNR, repeat_n), repeat_n_mod)\n",
        "\n",
        "    # X = tf.convert_to_tensor(X, dtype=tf.float32)\n",
        "    # one_hot = tf.convert_to_tensor(one_hot, dtype=tf.float32)\n",
        "\n",
        "    return SNRs, X, modulations, one_hot, lbl_SNR\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "zcYQebk3iXYq"
      },
      "source": [
        "# RadioML2016.10a/10b or MIGOU MOD\n",
        "\n",
        "def load_dataset(dataset_location):\n",
        "    \"\"\"\n",
        "    Load dataset and extract needed data\n",
        "\n",
        "    Input: \n",
        "        dataset_location: specify where the file is stored and its name\n",
        "\n",
        "    Output:\n",
        "        snrs: list of the SNR range in dataset [-20 to 18]\n",
        "        X: array of the measured I/Q data [num_of_samples, 128, 2]\n",
        "        modulations: list of the modulations in this dataset\n",
        "        one_hot_encode: one_hot encoded data - the other maps the order of the mods\n",
        "        lbl_SNR: list of each snr (for plotting)\n",
        "    \"\"\"\n",
        "\n",
        "    snrs,mods = map(lambda j: sorted(list(set(map(lambda x: x[j], dataset_location.keys())))), [1,0])\n",
        "\n",
        "    X = []; I = []; Q = []; lbl = [];\n",
        "\n",
        "    for mod in mods:\n",
        "        for snr in snrs:\n",
        "            X.append(dataset_location[(mod,snr)])\n",
        "            for i in range(dataset_location[(mod,snr)].shape[0]):  \n",
        "                lbl.append((mod,snr))\n",
        "    X = np.vstack(X); lbl=np.vstack(lbl)\n",
        "\n",
        "    X = np.transpose(X[:,:,:],(0,2,1))\n",
        "\n",
        "    # One-hot-encoding\n",
        "    Y = [];\n",
        "    for i in range(len(lbl)):\n",
        "        mod = (lbl[i,0])\n",
        "        Y.append(mod)\n",
        "\n",
        "    mapping = {}\n",
        "    for x in range(len(mods)):\n",
        "        mapping[mods[x]] = x\n",
        "\n",
        "    ## integer representation\n",
        "    for x in range(len(Y)):\n",
        "        Y[x] = mapping[Y[x]]\n",
        "\n",
        "    one_hot_encode = to_categorical(Y)\n",
        "\n",
        "    # Assign SNR value to each vector\n",
        "    repeat_n = X.shape[0]/len(mods)/len(snrs)\n",
        "    repeat_n_mod = len(mods)    \n",
        "    lbl_SNR = np.tile(np.repeat(snrs, repeat_n), repeat_n_mod)\n",
        "\n",
        "\n",
        "\n",
        "    return snrs, X, mods, one_hot_encode, lbl_SNR"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ej8VcSIpmn2g"
      },
      "source": [
        "# RML2016.10b / just for the way it is saved in my GoogleDrive\n",
        "\n",
        "def load_RMLb_dataset(X, lbl):\n",
        "    mods = np.unique(lbl[:,0])\n",
        "    snrs = np.unique(lbl[:,1])\n",
        "    snrs = list(map(int, snrs))\n",
        "    snrs.sort()\n",
        "\n",
        "    # One-hot encoding\n",
        "    Y = [];\n",
        "    for i in range(len(lbl)):\n",
        "        mod = (lbl[i,0])\n",
        "        Y.append(mod)\n",
        "\n",
        "    mapping = {}\n",
        "    for x in range(len(mods)):\n",
        "        mapping[mods[x]] = x\n",
        "\n",
        "    ## integer representation\n",
        "    for x in range(len(Y)):\n",
        "        Y[x] = mapping[Y[x]]\n",
        "\n",
        "    one_hot_encode = to_categorical(Y)\n",
        "\n",
        "\n",
        "    # Assign SNR value to each vector\n",
        "    repeat_n = X.shape[0]/len(mods)/len(snrs)\n",
        "    repeat_n_mod = len(mods)    \n",
        "    lbl_SNR = np.tile(np.repeat(snrs, repeat_n), repeat_n_mod)\n",
        "\n",
        "    X = X\n",
        "\n",
        "    return snrs, X, mods, one_hot_encode, lbl_SNR\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "4kTNvjBdeTC2"
      },
      "source": [
        "## Functions to handle the datasets"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "xSrPHN9hCv_Q"
      },
      "source": [
        "def train_test_valid_split(X, one_hot, train_split=0.7, valid_split=0.15, test_split=0.15):\n",
        "    \n",
        "    \"\"\"\n",
        "    Train-Test split the data\n",
        "\n",
        "    Input:\n",
        "        X: X data\n",
        "        one_hot: Y data encoded to one_hot\n",
        "        train_split (default 0.7)\n",
        "        valid_split (default 0.15)\n",
        "        test_split (default 0.15)\n",
        "        train_split : valid_split : test_split - ratio for splitting the dataset\n",
        "        \n",
        "        NOTE: the ratio split must be a sum of 1!\n",
        "\n",
        "    Output:\n",
        "        train_idx: indexes from X assinged to train data\n",
        "        valid_idx: indexes from X assinged to validation data \n",
        "        test_idx: indexes from X assinged to test data\n",
        "        X_train: X data assigned for training\n",
        "        X_valid: X data assigned for validation\n",
        "        X_test: X data assigned for testing\n",
        "        Y_train: one-hot encoded Y data assigned for training\n",
        "        Y_valid: one-hot encoded Y data assigned for validation\n",
        "        Y_test: one-hot encoded Y data assigned for testing\n",
        "    \"\"\"\n",
        "\n",
        "    # Set random seed\n",
        "    np.random.seed(42)\n",
        "    random.seed(42)\n",
        "\n",
        "    # Get the number of samples\n",
        "    n_examples = X.shape[0]\n",
        "    n_train = int(n_examples * train_split)\n",
        "    n_valid = int(n_examples * valid_split)\n",
        "    n_test = int(n_examples * test_split)\n",
        "    \n",
        "    # Get indexes of train data\n",
        "    train_idx = np.random.choice(range(0, n_examples), size=n_train, replace=False)\n",
        "\n",
        "    # Left indexes for valid and test sets\n",
        "    left_idx= list(set(range(0, n_examples)) - set(train_idx))\n",
        "    \n",
        "    # Get indexes for the left indexes of the X data\n",
        "    val = np.random.choice(range(0, (n_valid+n_test)), size=(n_valid), replace=False)\n",
        "    test = list(set(range(0, len(left_idx))) - set(val))\n",
        "\n",
        "    # Assign indeces for validation to left indexes\n",
        "    valid_idx = []\n",
        "    for i in val:\n",
        "        val_idx = left_idx[i]\n",
        "        valid_idx.append(val_idx)\n",
        "    \n",
        "    # Get the test set as the rest indexes\n",
        "    test_idx = []\n",
        "    for i in test:\n",
        "        tst_idx = left_idx[i]\n",
        "        test_idx.append(tst_idx)\n",
        "    \n",
        "    # Shuffle the valid_idx and test_idx\n",
        "    random.shuffle(valid_idx)\n",
        "    random.shuffle(test_idx)\n",
        "\n",
        "    # Assing the indexes to the X and Y data to create train and test sets\n",
        "    X_train = X[train_idx]\n",
        "    X_valid = X[valid_idx]\n",
        "    X_test = X[test_idx]   \n",
        "    Y_train = one_hot[train_idx]\n",
        "    Y_valid = one_hot[valid_idx]\n",
        "    Y_test = one_hot[test_idx]  \n",
        "\n",
        "    return train_idx, valid_idx, test_idx, X_train, X_valid, X_test, Y_train, Y_valid, Y_test\n",
        "\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "5VLh2-YxlV_9"
      },
      "source": [
        "def normalize_data(X_train, X_valid, X_test):\n",
        "        # mean-std normalization\n",
        "\n",
        "    mean = X_train[:,:,:].mean(axis=0)\n",
        "    X_train[:,:,:] -= mean\n",
        "    std = X_train[:,:,:].std(axis=0)\n",
        "    X_train[:,:,:] /= std\n",
        "\n",
        "\n",
        "    X_valid[:,:,:] -= mean\n",
        "    X_valid[:,:,:] /= std\n",
        "\n",
        "    X_test[:,:,:] -= mean\n",
        "    X_test[:,:,:] /= std\n",
        "\n",
        "    return X_train, X_valid, X_test"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "0LGV76g1dhdy"
      },
      "source": [
        "def return_indices_of_a(a, b):\n",
        "    \"\"\"\n",
        "    Compare two lists a, b for same items and return indeces\n",
        "    of the item in list a\n",
        "\n",
        "    a:    List of items, its indeces will be returned\n",
        "    b:    List of items to search for in list a\n",
        "\n",
        "    Credit: https://stackoverflow.com/users/97248/pts ; https://stackoverflow.com/questions/10367020/compare-two-lists-in-python-and-return-indices-of-matched-values\n",
        "    \"\"\"\n",
        "    b_set = set(b)\n",
        "    return [i for i, v in enumerate(a) if v in b_set]"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "6GYK_g31rtAl"
      },
      "source": [
        "## Functions for plotting"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "kQTTjxsRr0uu"
      },
      "source": [
        "def show_confusion_matrix(validations, predictions, matrix_snr, save=False):\n",
        "    \"\"\"\n",
        "    Plot confusion matrix\n",
        "\n",
        "    validations:    True Y labels\n",
        "    predictions:    Predicted Y labels of your model\n",
        "    matrix_snr:     SNR information for plot's titel\n",
        "    \"\"\"\n",
        "  \n",
        "    cm = confusion_matrix(validations, predictions)\n",
        "    # Normalise\n",
        "    cmn = cm.astype('float') / cm.sum(axis=1)[:, np.newaxis]\n",
        "    fig, ax = plt.subplots(figsize=(10,10))\n",
        "    sns.heatmap(cmn, cmap='Blues', annot=True, fmt='.2f', xticklabels=mods, yticklabels=mods)\n",
        "    sns.set(font_scale=1.3)\n",
        "    if matrix_snr == None:\n",
        "        plt.title(\"Confusion Matrix\")\n",
        "    else:\n",
        "        plt.title(\"Confusion Matrix \\n\" + str(matrix_snr) + \"dB\")\n",
        "    plt.ylabel('True Label')\n",
        "    plt.xlabel('Predicted Label')\n",
        "    if save == True:\n",
        "        plt.savefig(base_dir + 'Own_dataset/' + str(matrix_snr) + '.png')    \n",
        "    plt.show(block=False)\n",
        "    "
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "GyTrNPxjrk4Y"
      },
      "source": [
        "def All_SNR_show_confusion_matrix(X_test, save=False):\n",
        "    \"\"\"\n",
        "    Plot confusion matrix of all SNRs in one\n",
        "\n",
        "    X_test:   X_test data\n",
        "    \"\"\"\n",
        "    prediction = model.predict(X_test)\n",
        "\n",
        "    Y_Pred = []; Y_Test = [];\n",
        "\n",
        "    for i in range(len(prediction[:,0])):\n",
        "        Y_Pred.append(np.argmax(prediction[i,:]))\n",
        "        Y_Test.append(np.argmax(Y_test[i]))\n",
        "\n",
        "    show_confusion_matrix(Y_Pred, Y_Test, None, save)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "awOLhlX3JshY"
      },
      "source": [
        "def SNR_show_confusion_matrix(in_snr, lbl_SNR, X_test, save=False):\n",
        "    \"\"\"\n",
        "    Plot confusion matrices of chosen SNRs\n",
        "\n",
        "    in_snr:   must be list of SNRs\n",
        "    X_test:   X_test data\n",
        "    \"\"\"\n",
        "    for snr in in_snr:\n",
        "        matrix_snr = snr\n",
        "        m_snr = matrix_snr;\n",
        "\n",
        "        Y_Pred = []; Y_Test = []; Y_Pred_SNR = []; Y_Test_SNR = []; \n",
        "        matrix_snr_index = [];\n",
        "\n",
        "        prediction = model.predict(X_test)\n",
        "\n",
        "        for i in range(len(prediction[:,0])):\n",
        "            Y_Pred.append(np.argmax(prediction[i,:]))\n",
        "            Y_Test.append(np.argmax(Y_test[i]))\n",
        "\n",
        "        for i in range(len(lbl_SNR)):\n",
        "            if int(lbl_SNR[i]) == m_snr:\n",
        "                matrix_snr_index.append(i)\n",
        "\n",
        "        indeces_of_Y_test = return_indices_of_a(test_idx, matrix_snr_index)\n",
        "\n",
        "        for i in indeces_of_Y_test:\n",
        "            Y_Pred_SNR.append(Y_Pred[i])\n",
        "            Y_Test_SNR.append(Y_Test[i])\n",
        "        show_confusion_matrix(Y_Pred_SNR, Y_Test_SNR, matrix_snr, save)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "_r1XU8jrHjSF"
      },
      "source": [
        "def plot_split_distribution(mods, Y_train, Y_valid, Y_test):\n",
        "\n",
        "    x = np.arange(len(mods))  # the label locations\n",
        "    width = 1  # the width of the bars\n",
        "\n",
        "    fig, ax = plt.subplots()\n",
        "    bar1 = ax.bar(x-width*0.3, np.count_nonzero(Y_train == 1, axis=0), width*0.3, label = \"Train\" )\n",
        "    bar2 = ax.bar(x , np.count_nonzero(Y_valid == 1, axis=0), width*0.3, label = \"Valid\" )\n",
        "    bar3 = ax.bar(x+width*0.3, np.count_nonzero(Y_test == 1, axis=0), width*0.3, label = \"Test\" )\n",
        "\n",
        "\n",
        "    # Add some text for labels, title and custom x-axis tick labels, etc.\n",
        "    ax.set_ylabel('Distribution')\n",
        "    ax.set_title('Distribution overview of splitted dataset')\n",
        "    ax.set_xticks(x)\n",
        "    ax.set_xticklabels(mods)\n",
        "    ax.legend(loc='upper center', bbox_to_anchor=(0.5, -0.15),\n",
        "            fancybox=True, shadow=True, ncol=5)\n",
        "\n",
        "\n",
        "    def autolabel(rects):\n",
        "        \"\"\"Attach a text label above each bar in *rects*, displaying its height.\"\"\"\n",
        "        for rect in rects:\n",
        "            height = rect.get_height()\n",
        "            ax.annotate('{}'.format(height),\n",
        "                        xy=(rect.get_x() + rect.get_width() / 2, height),\n",
        "                        xytext=(0, 0),  # 3 points vertical offset\n",
        "                        textcoords=\"offset points\",\n",
        "                        ha='center', va='bottom')\n",
        "            \n",
        "    # autolabel(bar1)\n",
        "    # autolabel(bar2)\n",
        "    # autolabel(bar3)\n",
        "    # fig.tight_layout()\n",
        "    return plt.show()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Jm5KnGukcnSs"
      },
      "source": [
        "def SNR_accuracy(in_snr, name):\n",
        "    \"\"\"\n",
        "    Computes accuracies of chosen SNRs individualy\n",
        "  \n",
        "    in_snr:   must be list of SNRs\n",
        "    \"\"\"\n",
        "    \n",
        "    acc = []\n",
        "    for snr in in_snr:\n",
        "        acc_snr = snr\n",
        "        idx_acc_snr = []\n",
        "\n",
        "        for i in range(len(test_idx)):\n",
        "            if int(lbl_SNR[test_idx[i]]) == int(acc_snr):\n",
        "                idx_acc_snr.append(i)\n",
        "\n",
        "        acc_X_test = X_test[idx_acc_snr]\n",
        "        # acc_X_f_test = X_f_test[idx_acc_snr]\n",
        "        acc_Y_test = Y_test[idx_acc_snr]\n",
        "\n",
        "        print('\\nSNR ' + str(acc_snr) + 'dB:')\n",
        "        accuracy_snr = model.evaluate([acc_X_test], acc_Y_test, batch_size=32, verbose=2)\n",
        "        acc.append(accuracy_snr)\n",
        "\n",
        "    acc = np.vstack(acc)\n",
        "    fig = plt.figure()\n",
        "    ax = fig.add_subplot(1, 1, 1)\n",
        "    plt.plot(SNR, (acc[:,1]*100), 'steelblue', marker='.', markersize= 15, label = name, linestyle = '-',)\n",
        "    ax.legend(loc=4, prop={'size': 25})\n",
        "\n",
        "    x_major_ticks = np.arange(-20, 19, 2 )\n",
        "    ax.set_xticks(x_major_ticks)\n",
        "\n",
        "    y_major_ticks = np.arange(0, 101, 10 )\n",
        "    y_minor_ticks = np.arange(0, 101, 2)\n",
        "    ax.set_yticks(y_major_ticks)\n",
        "    ax.set_yticks(y_minor_ticks, minor=True)\n",
        "    ax.tick_params(axis='both', which='major', labelsize=20)\n",
        "\n",
        "    ax.grid(which='both',color='lightgray', linestyle='-')\n",
        "\n",
        "    ax.grid(which='minor', alpha=0.2)\n",
        "    ax.grid(which='major', alpha=0.5)\n",
        "\n",
        "    plt.xlim(-20, 18)\n",
        "    plt.ylim(0,100)\n",
        "    plt.title(\"Classification Accuracy\",fontsize=20)\n",
        "    plt.ylabel('Accuracy (%)',fontsize=20)\n",
        "    plt.xlabel('SNR (dB)',fontsize=20)\n",
        "    # plt.savefig(base_dir + name + '.png') \n",
        "    plt.show()\n",
        "\n",
        "    return acc[:,1]\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "vs1M0dXbEkvf"
      },
      "source": [
        "## Functions for visualization of layers"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "APgVxDHuEq5Z"
      },
      "source": [
        "def layer_overview(model):\n",
        "\n",
        "    \"\"\"\n",
        "    Offers overview of the model's layers and theirs outputs\n",
        "\n",
        "    model: specify trained model you want to have overview of\n",
        "    \"\"\"\n",
        "\n",
        "    # Names and outputs from layers\n",
        "    layer_names = [layer.name for layer in model.layers]\n",
        "    layer_outputs = [layer.output for layer in model.layers[:]]\n",
        "\n",
        "    return layer_names, layer_outputs\n",
        "\n",
        "def model_visualization(nth_layer, nth_test_idx, mods, model,\n",
        "                        plot_sample = False, plot_activations = True, \n",
        "                        plot_feature_maps = True):\n",
        "    \n",
        "    \"\"\"\n",
        "    The function provised overview of activation of specific layer and its\n",
        "    feature maps.\n",
        "\n",
        "    nth_layer: enter number which corresponds with the position of wanted layer \n",
        "    nth_test_idx: enter number pointing at the test indexes from earlier\n",
        "    mods: provide variable which holds listed modulations\n",
        "    model: specify which trained model to load\n",
        "    plot_sample = False: set to true to plot sample data\n",
        "    plot_activations = True: plots activation of chosen layer\n",
        "    plot_feature_maps = True: plots feature map of chosen layer\n",
        "    \"\"\"\n",
        "\n",
        "    # Sample data for visualization\n",
        "    test_sample = X_test[nth_test_idx,:,:] # shape [128,2]\n",
        "    test_sample = test_sample[None] # change to needed [1,128,2]\n",
        "    SNR = lbl_SNR[test_idx[nth_test_idx]]\n",
        "    mod = one_hot[test_idx[nth_test_idx]]\n",
        "    f, u = factorize(mods)\n",
        "    mod = mod.dot(u)\n",
        "\n",
        "    # Names and outputs from layers\n",
        "    layer_names = [layer.name for layer in model.layers]\n",
        "    layer_outputs = [layer.output for layer in model.layers[:]]\n",
        "\n",
        "    ## Activations ##\n",
        "\n",
        "    # define activation model\n",
        "    activation_model = tf.keras.models.Model(model.input, layer_outputs)\n",
        "\n",
        "    # get the activations of chosen test sample\n",
        "    activations = activation_model.predict(test_sample)\n",
        "\n",
        "    ## Feature-maps ##\n",
        "\n",
        "    # define feature maps model \n",
        "    feature_maps_model = tf.keras.models.Model(model.inputs, model.layers[4].output)\n",
        "    \n",
        "    # get the activated features\n",
        "    feature_maps = feature_maps_model.predict(test_sample)\n",
        "\n",
        "\n",
        "    # Plot sample\n",
        "    if plot_sample == True:\n",
        "        plt.plot(test_sample[0,:,:])\n",
        "        plt.title(mod + '    ' + str(SNR) + 'dB')\n",
        "        plt.show()\n",
        "\n",
        "    # Plot activations\n",
        "    if plot_activations == True:\n",
        "        activation_layer = activations[nth_layer]\n",
        "        activation_layer = np.transpose(activation_layer[:,:,:],(0,2,1)) # reshape\n",
        "        fig, ax = plt.subplots(figsize=(20,10))\n",
        "        ax.matshow(activation_layer[0,:,:],  cmap='viridis')\n",
        "        # plt.matshow(activation_layer[0,:,:],  cmap='viridis')\n",
        "        plt.title('Activation of layer ' + layer_names[nth_layer])\n",
        "        ax.grid(False)\n",
        "        ax.set_xlabel('Lenght of sequence')\n",
        "        ax.set_ylabel('Filters')\n",
        "        fig.show()\n",
        "        plt.savefig(base_dir + 'activations.png')\n",
        "        plt.savefig(base_dir + 'activations.svg')\n",
        "\n",
        "    # Plot feature maps\n",
        "    if plot_feature_maps == True:\n",
        "        n_filters = int(feature_maps.shape[2]/2); ix = 1\n",
        "        fig = plt.figure(figsize=(25,15))\n",
        "        for _ in range(n_filters):\n",
        "            for _ in range(2):\n",
        "                # specify subplot and turn of axis\n",
        "                ax =fig.add_subplot(n_filters,  5, ix)\n",
        "                # ax = plt.subplot(n_filters,  5, ix, )\n",
        "                ax.set_xticks([])\n",
        "                ax.set_yticks([])\n",
        "                # plot filter channel in grayscale\n",
        "                ax.plot(feature_maps[0, :, ix-1])\n",
        "                ix += 1\n",
        "        # show the figure\n",
        "        fig.show()\n",
        "        plt.savefig(base_dir + 'feature_map.png')\n",
        "        plt.savefig(base_dir + 'feature_map.svg')"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "3agW4t79sXqP"
      },
      "source": [
        "## Transformer"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "bALyj-oqsk11"
      },
      "source": [
        "def position_encoding_init(n_position, emb_dim):\n",
        "    ''' Init the sinusoid position encoding table '''\n",
        "\n",
        "    # keep dim 0 for padding token position encoding zero vector\n",
        "    position_enc = np.array([\n",
        "        [pos / np.power(10000, 2 * (j // 2) / emb_dim) for j in range(emb_dim)]\n",
        "        if pos != 0 else np.zeros(emb_dim) for pos in range(n_position)])\n",
        "    \n",
        "\n",
        "    position_enc[1:, 0::2] = np.sin(position_enc[1:, 0::2]) # dim 2i\n",
        "    position_enc[1:, 1::2] = np.cos(position_enc[1:, 1::2]) # dim 2i+1\n",
        "    \n",
        "    return position_enc"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "uymqyGEn9un2"
      },
      "source": [
        "# Transformer Block\n",
        "class TransformerBlock(layers.Layer):\n",
        "    def __init__(self, embed_dim, num_heads, ff_dim, rate=0.1):\n",
        "        super(TransformerBlock, self).__init__()\n",
        "        self.att = layers.MultiHeadAttention(num_heads=num_heads, key_dim=embed_dim)\n",
        "        self.ffn = keras.Sequential(\n",
        "            [layers.Dense(ff_dim, activation=\"relu\"), layers.Dense(embed_dim),]\n",
        "        )\n",
        "        self.layernorm1 = layers.LayerNormalization(epsilon=1e-6)\n",
        "        self.layernorm2 = layers.LayerNormalization(epsilon=1e-6)\n",
        "        self.dropout1 = layers.Dropout(rate)\n",
        "        self.dropout2 = layers.Dropout(rate)\n",
        "\n",
        "    def call(self, inputs, training):\n",
        "        attn_output = self.att(inputs, inputs)\n",
        "        attn_output = self.dropout1(attn_output, training=training)\n",
        "        out1 = self.layernorm1(inputs + attn_output)\n",
        "        ffn_output = self.ffn(out1)\n",
        "        ffn_output = self.dropout2(ffn_output, training=training)\n",
        "        return self.layernorm2(out1 + ffn_output)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "pn_6aSoKM_SH"
      },
      "source": [
        "# Access the datasets\n",
        "\n",
        "With the following cells, you can easily access the datasets. However, if you end up using them for your work, do not forget to credit the original authors! More info is provided for each of them below."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "L8dfXIAIOSl-"
      },
      "source": [
        "# Uncomment the following line, if needed, to download the datasets\n",
        "# !conda install -y gdown"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "puzzG2bMPB7z"
      },
      "source": [
        "## RadioML Datasets\n",
        "\n",
        "\n",
        "*  O'shea, Timothy J., and Nathan West. \"Radio machine learning dataset generation with gnu radio.\" Proceedings of the GNU Radio Conference. Vol. 1. No. 1. 2016.\n",
        "\n",
        "* The datasets are available at:  https://www.deepsig.ai/datasets  \n",
        "\n",
        "*  All datasets provided by Deepsig Inc. are licensed under the Creative Commons Attribution -  [NonCommercial - ShareAlike 4.0 License (CC BY-NC-SA 4.0)](https://creativecommons.org/licenses/by-nc-sa/4.0/).\n",
        "\n",
        "Both datasets are left unchanged, however, the RadioML2016.10b version is not stored as the original data, but is already splitted into X and labels\n"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "x2u1KR29Oe1B"
      },
      "source": [
        "# RadioML2016.10a stored as the original pkl file\n",
        "!gdown --id 1aus-u2xSKETW9Yv5Q-QG9tz9Xnbj5yHV"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "1zfnS8Z2SQMi"
      },
      "source": [
        "dataset_pkl = open('RML2016.10a_dict.pkl','rb')\n",
        "RML_dataset_location = pickle.load(dataset_pkl, encoding='bytes')"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "KV_Vm6b-NB_P"
      },
      "source": [
        "# RadioML2016.10b stored in X.pkl and label.pkl\n",
        "!gdown --id 10OdxNvtSbOm58t-MMHZcmSMqzEWDSpAr\n",
        "!gdown --id 1-MvVKNmTfqyfYD_usvAfEcizzBX0eEpE"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "gQfuzAzfSSgu"
      },
      "source": [
        "RMLb_X_data_file = open('X.pkl','rb')\n",
        "RMLb_labels_file = open('labels.pkl', 'rb')\n",
        "RMLb_X = pickle.load(RMLb_X_data_file, encoding='bytes')\n",
        "RMLb_lbl = pickle.load(RMLb_labels_file, encoding='ascii') "
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_qg5EnYDQlj5"
      },
      "source": [
        "## Migou-Mod Dataset\n",
        "\n",
        "\n",
        "*  Utrilla, Ramiro (2020), “MIGOU-MOD: A dataset of modulated radio signals acquired with MIGOU, a low-power IoT experimental platform”, Mendeley Data, V1, doi: 10.17632/fkwr8mzndr.1\n",
        "\n",
        "* The dataset is available at:  https://data.mendeley.com/datasets/fkwr8mzndr/1 \n",
        "\n",
        "*  The dataset is licensed under the Creative Commons Attribution -  [NonCommercial - ShareAlike 4.0 License (CC BY-NC-SA 4.0)](https://creativecommons.org/licenses/by-nc-sa/4.0/).\n",
        "\n",
        "The following version of the dataset contain only a fraction of the original samples (550,000 samples compared to 8.8 million samples in the original dataset)\n"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ns7HjFzNOPc8"
      },
      "source": [
        "# Migou-Mod Dataset - 550,000 samples \n",
        "!gdown --id 1-CIL3bD4o9ylBkD0VZkGd5n1-8_RTRvs"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "tBBOb-CiSVpb"
      },
      "source": [
        "MIGOU_dataset_pkl = open('dataset_25.pkl','rb')\n",
        "MIGOU_dataset_location = pickle.load(MIGOU_dataset_pkl, encoding='bytes')"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "0ww77k0pRZrp"
      },
      "source": [
        "## VUT Dataset\n",
        "\n",
        "This dataset was generated in MATLAB with 1000 samples per SNR value and each modulation type. It includes three QAM modulation schemes and further OFDM, GFDM, and FBMC modulations which are not included in previous datasets. To mimic the RadioML dataset, the data are represented as 2x128 vectors of I/Q signals in the SNR range from -20 dB to 18 dB."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "d9DJ369SONaX"
      },
      "source": [
        "# VUT Dataset\n",
        "!gdown --id 1G5WsgUze8qfuSzy6Edg_4qRIiAx_YUc4"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "OjRYr6ixSXoc"
      },
      "source": [
        "VUT_dataset_location = 'NEW_Dataset_05_02_2021.mat'"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "wU6Mg8rNxVRf"
      },
      "source": [
        "# Load the data"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "de4qTdA_hJKj"
      },
      "source": [
        "## VUT Dataset\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "0qAox-RM_WE5"
      },
      "source": [
        "SNR, X, mods, one_hot, lbl_SNR = load_VUT_dataset(VUT_dataset_location)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "sSrhWh9OS7R7"
      },
      "source": [
        "train_idx, valid_idx, test_idx, X_train, X_valid, X_test, Y_train, Y_valid, Y_test = train_test_valid_split(X, one_hot, train_split=0.7, valid_split=0.15, test_split=0.15)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "_kduOR1HQ4DN"
      },
      "source": [
        "plot_split_distribution(mods, Y_train, Y_valid, Y_test)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "sA1QnRJ5aNIU"
      },
      "source": [
        "## DeepSig Dataset"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "tn-UlZPXhwQo"
      },
      "source": [
        "# 10a\n",
        "# SNR, X, modulations, one_hot, lbl_SNR = load_dataset(RML_dataset_location)\n",
        "\n",
        "# 10b\n",
        "SNR, X, modulations, one_hot, lbl_SNR = load_RMLb_dataset(RMLb_X, RMLb_lbl)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "7tBdh7OCnKR6"
      },
      "source": [
        "mods = []\n",
        "for i in range(len(modulations)):\n",
        "    modu = modulations[i].decode('utf-8')\n",
        "    mods.append(modu)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "vWHy1i4lbxhn"
      },
      "source": [
        "train_idx, valid_idx, test_idx, X_train, X_valid, X_test, Y_train, Y_valid, Y_test = train_test_valid_split(X, one_hot, train_split=0.7, valid_split=0.15, test_split=0.15)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "eMA1rRIccDvo"
      },
      "source": [
        "plot_split_distribution(mods, Y_train, Y_valid, Y_test)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "bu_HQdJucCe0"
      },
      "source": [
        "# X_train, X_valid, X_test = normalize_data(X_train, X_valid, X_test)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "S3tzTIiCyHd8"
      },
      "source": [
        "## MIGOU-MOD"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "5by1HAyCjIKN"
      },
      "source": [
        "SNR, X, mods, one_hot, lbl_SNR = load_dataset(MIGOU_dataset_location)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "jBGnpgTaSSpF"
      },
      "source": [
        "train_idx, valid_idx, test_idx, X_train, X_valid, X_test, Y_train, Y_valid, Y_test = train_test_valid_split(X, one_hot, train_split=0.7, valid_split=0.15, test_split=0.15)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "gZx1uTHm7KbZ"
      },
      "source": [
        "plot_split_distribution(mods, Y_train, Y_test, Y_test)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "z24B4-75qu2N"
      },
      "source": [
        "# Architectures for training"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "cQyeFq49fICR"
      },
      "source": [
        "## CNN"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "1ToHFfS5qQ8i"
      },
      "source": [
        "\n",
        "cnn_in = keras.layers.Input(shape=(128,2))\n",
        "cnn = keras.layers.ZeroPadding1D(padding=4)(cnn_in)\n",
        "cnn = keras.layers.Conv1D(filters=50, kernel_size=8, activation='relu')(cnn)\n",
        "cnn = keras.layers.MaxPool1D(pool_size=2)(cnn)\n",
        "cnn = keras.layers.Conv1D(filters=50, kernel_size=8, activation='relu')(cnn)\n",
        "cnn = keras.layers.MaxPool1D(pool_size=2)(cnn)\n",
        "cnn = keras.layers.Conv1D(filters=50, kernel_size=4, activation='relu')(cnn)\n",
        "cnn = keras.layers.Dropout(rate=0.6)(cnn)\n",
        "cnn = keras.layers.MaxPool1D(pool_size=2)(cnn)\n",
        "cnn = keras.layers.Flatten()(cnn)\n",
        "cnn = keras.layers.Dense(70, activation='selu')(cnn)\n",
        "cnn_out = keras.layers.Dense(len(mods), activation='softmax')(cnn)\n",
        "\n",
        "model_cnn = keras.models.Model(cnn_in, cnn_out)\n",
        "\n",
        "callbacks = [\n",
        "    keras.callbacks.ModelCheckpoint(\n",
        "        \"cnn_model.h5\", save_best_only=True, monitor=\"val_loss\"),\n",
        "    keras.callbacks.ReduceLROnPlateau(\n",
        "        monitor=\"val_loss\", factor=0.3, patience=3, min_lr=0.00007),\n",
        "    keras.callbacks.EarlyStopping(monitor=\"val_loss\", patience=5, verbose=1)]\n",
        "\n",
        "optimizer = keras.optimizers.Adam(learning_rate=0.0007)\n",
        "\n",
        "model_cnn.compile(loss='categorical_crossentropy', optimizer=optimizer, metrics=['accuracy'])"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "PaEbCljq8MV-"
      },
      "source": [
        "# model_cldnn.summary()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "9esNkdTkssQE"
      },
      "source": [
        "tf.keras.backend.clear_session()\n",
        "history = model_cnn.fit(X_train, Y_train, batch_size=128, epochs=4, verbose=2, validation_data= (X_valid, Y_valid), callbacks=callbacks)\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "JzpSswHtsuVJ"
      },
      "source": [
        "model = keras.models.load_model(\"cnn_model.h5\")\n",
        "\n",
        "test_loss, test_acc = model.evaluate(X_test, Y_test)\n",
        "\n",
        "print(\"Test accuracy\", test_acc)\n",
        "print(\"Test loss\", test_loss)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "xf_H7YN7f6gY"
      },
      "source": [
        "SNR_accuracy(SNR, 'CNN')"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "1b0RhtKtfMMT"
      },
      "source": [
        "## CLDNN"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "3z0pnIOmtB6F"
      },
      "source": [
        "layer_in = keras.layers.Input(shape=(128,2))\n",
        "layer = keras.layers.Conv1D(filters=64, kernel_size=8, activation='relu')(layer_in)\n",
        "layer = keras.layers.MaxPool1D(pool_size=2)(layer)\n",
        "layer = keras.layers.LSTM(64, return_sequences=True,)(layer)\n",
        "layer = keras.layers.Dropout(0.4)(layer)\n",
        "layer = keras.layers.LSTM(64, return_sequences=True,)(layer)\n",
        "layer = keras.layers.Dropout(0.4)(layer)\n",
        "layer = keras.layers.Flatten()(layer)\n",
        "layer_out = keras.layers.Dense(len(mods), activation='softmax')(layer)\n",
        "\n",
        "model_cldnn = keras.models.Model(layer_in, layer_out)\n",
        "\n",
        "optimizer = keras.optimizers.Adam(learning_rate=0.0007)\n",
        "\n",
        "callbacks = [\n",
        "    keras.callbacks.ModelCheckpoint(\n",
        "        \"cldnn_model.h5\", save_best_only=True, monitor=\"val_loss\"),\n",
        "    keras.callbacks.ReduceLROnPlateau(\n",
        "        monitor=\"val_loss\", factor=0.4, patience=5, min_lr=0.000007),\n",
        "    keras.callbacks.EarlyStopping(monitor=\"val_loss\", patience=8, verbose=1)]\n",
        "\n",
        "model_cldnn.compile(loss='categorical_crossentropy', optimizer=optimizer, metrics=['accuracy'])"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "M8yXuNBd1qzb"
      },
      "source": [
        "# model_cldnn.summary()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "5xXo6Iu5tNo_"
      },
      "source": [
        "tf.keras.backend.clear_session()\n",
        "history = model_cldnn.fit(X_train, Y_train, batch_size=128, epochs=100, verbose=2, validation_data= (X_valid, Y_valid), callbacks=callbacks)\n",
        "# history = model_iq.fit(X_train, Y_train, batch_size=128, epochs=100, verbose=2, validation_split=0.15, callbacks=callbacks)\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "BVka0eS-vSAt"
      },
      "source": [
        "model = keras.models.load_model(\"cldnn_model.h5\")\n",
        "\n",
        "test_loss, test_acc = model.evaluate(X_test, Y_test)\n",
        "\n",
        "print(\"Test accuracy\", test_acc)\n",
        "print(\"Test loss\", test_loss)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "zrxRxbQvjI9a"
      },
      "source": [
        "SNR_accuracy(SNR, 'CLDNN')"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "jLaqnWAPfkwB"
      },
      "source": [
        "## GGDNN"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "_RmG8wrQPOl3"
      },
      "source": [
        "layer_in = keras.layers.Input(shape=(128,2)) \n",
        "layer = keras.layers.Conv1D(filters=80, kernel_size=(12), activation='relu')(layer_in)\n",
        "layer = keras.layers.MaxPool1D(pool_size=(2))(layer)\n",
        "layer = keras.layers.GRU(40, return_sequences=True)(layer)\n",
        "layer = keras.layers.GaussianDropout(0.4)(layer)\n",
        "layer = keras.layers.GRU(40, return_sequences=True)(layer)\n",
        "layer = keras.layers.GaussianDropout(0.4)(layer)\n",
        "layer = keras.layers.Flatten()(layer)\n",
        "layer_out = keras.layers.Dense(10, activation='softmax')(layer)\n",
        "\n",
        "model_CGDNN = keras.models.Model(layer_in, layer_out)\n",
        "\n",
        "optimizer = keras.optimizers.Adam(learning_rate=0.002)\n",
        "\n",
        "callbacks = [\n",
        "    keras.callbacks.ModelCheckpoint(\n",
        "        \"cgdnn_model.h5\", save_best_only=True, monitor=\"val_loss\"),\n",
        "    keras.callbacks.ReduceLROnPlateau(\n",
        "        monitor=\"val_loss\", factor=0.4, patience=4, min_lr=0.000007),\n",
        "    keras.callbacks.EarlyStopping(monitor=\"val_loss\", patience=10, verbose=1)]\n",
        "\n",
        "model_CGDNN.compile(loss='categorical_crossentropy', optimizer=optimizer, metrics=['accuracy'])"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "OcGPtivOPf-V"
      },
      "source": [
        "# model_CGDNN.summary()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "8q19f1eNPUy6"
      },
      "source": [
        "tf.keras.backend.clear_session()\n",
        "history = model_CGDNN.fit(X_train, Y_train, batch_size=128, epochs=100, verbose=2, validation_data=(X_valid,Y_valid), callbacks=callbacks)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "mYS8S6dyQcxq"
      },
      "source": [
        "model = keras.model_CGDNN.load_model(\"cgdnn_model.h5\")\n",
        "\n",
        "test_loss, test_acc = model.evaluate(X_test, Y_test)\n",
        "\n",
        "print(\"Test accuracy\", test_acc)\n",
        "print(\"Test loss\", test_loss)\n",
        "\n",
        "SNR_accuracy(SNR, 'CLGDNN')"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "nTXu68APfhdb"
      },
      "source": [
        "## MCTransformer"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "oGaEZ0wZyT9a"
      },
      "source": [
        "embed_dim = 64  # Embedding size for each token\n",
        "num_heads = 4  # Number of attention heads\n",
        "ff_dim = 16  # Hidden layer size in feed forward network inside transformer\n",
        "\n",
        "inputs = keras.layers.Input(shape=(128,2))\n",
        "x = keras.layers.Conv1D(filters=embed_dim, kernel_size=8, activation='relu')(inputs)\n",
        "x = keras.layers.MaxPool1D(pool_size=2)(x)\n",
        "x = keras.layers.LSTM(embed_dim, return_sequences=True,)(x)\n",
        "x = keras.layers.Dropout(0.4)(x)\n",
        "\n",
        "pos_emb = position_encoding_init(60,64)\n",
        "x_pos = x+pos_emb\n",
        "\n",
        "transformer_block = TransformerBlock(embed_dim, num_heads, ff_dim)\n",
        "x = transformer_block(x_pos)\n",
        "\n",
        "x = layers.GlobalAveragePooling1D()(x)\n",
        "x = layers.Dropout(0.1)(x)\n",
        "x = layers.Dense(20, activation=\"relu\")(x)\n",
        "x = layers.Dropout(0.1)(x)\n",
        "outputs = layers.Dense(len(mods), activation=\"softmax\")(x)\n",
        "\n",
        "model_MCT = keras.Model(inputs=inputs, outputs=outputs)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ARajb4PUInEN"
      },
      "source": [
        "# model_MCT.summary()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "VwETIzghMVlF"
      },
      "source": [
        "optimizer = keras.optimizers.SGD(learning_rate=0.03)\n",
        "model_MCT.compile(loss='categorical_crossentropy', optimizer=optimizer, metrics=['accuracy'])"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "yRxkgoitugGs"
      },
      "source": [
        "history = model_MCT.fit(X_train, Y_train, batch_size=16, epochs=20, validation_data= (X_valid, Y_valid))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "kGzV1Zle-4oz"
      },
      "source": [
        "Uncomment and lower the learning rate, if the validation loss doesn't improve."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "tQa2L3Dy3E6v"
      },
      "source": [
        "# optimizer = keras.optimizers.SGD(learning_rate=0.01)\n",
        "# model.compile(loss='categorical_crossentropy', optimizer=optimizer, metrics=['accuracy'])\n",
        "# history = model.fit(X_train, Y_train, batch_size=16, epochs=10, validation_data= (X_valid, Y_valid))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "DmHxMD5wIFAx"
      },
      "source": [
        "# optimizer = keras.optimizers.SGD(learning_rate=0.005)\n",
        "# model.compile(loss='categorical_crossentropy', optimizer=optimizer, metrics=['accuracy'])\n",
        "# history = model.fit(X_train, Y_train, batch_size=16, epochs=10, validation_data= (X_valid, Y_valid))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "fd38zDYSGoHa"
      },
      "source": [
        "# optimizer = keras.optimizers.SGD(learning_rate=0.001)\n",
        "# model.compile(loss='categorical_crossentropy', optimizer=optimizer, metrics=['accuracy'])\n",
        "# history = model.fit(X_train, Y_train, batch_size=128, epochs=10, validation_data= (X_valid, Y_valid))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "NsE6CIy7Xg96"
      },
      "source": [
        "test_loss, test_acc = model_MCT.evaluate(X_test, Y_test)\n",
        "\n",
        "print(\"Test accuracy\", test_acc)\n",
        "print(\"Test loss\", test_loss)\n",
        "SNR_accuracy(SNR, 'MCT')"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "3V0jZtyLiOqF"
      },
      "source": [
        "# Load saved CGDNN models"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "blosH6c2dpjU"
      },
      "source": [
        "Download the models"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "WllwwJ0vb7V9"
      },
      "source": [
        "# # RadioML2016.10a\n",
        "\n",
        "# !gdown --id 1h0iVzR0qEPEwcUEPKM3hBGF46uXQEs_l\n",
        "\n",
        "# # RadioML2016.10b\n",
        "# !gdown --id 1XCPOHF8ZeSC61qR1hrFKhgUxPHbpHg6R\n",
        "\n",
        "# # Migou-Mod Dataset\n",
        "# !gdown --id 1s4Uz5KlkLVO9lQyrJwVTW_754RNkigoC\n",
        "\n",
        "# # VUT Dataset\n",
        "# !gdown --id 1DWr1uDzz7m7rEfcKWXZXJpJ692EC0vBw"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "YEOnt2Q9dr_M"
      },
      "source": [
        "Uncomment wanted model \n",
        "\n",
        "Don't forget you also need to load the right dataset before predicting "
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "W7c9XzaVcq37"
      },
      "source": [
        "# RadioML2016.10a\n",
        "# model = tf.keras.models.load_model(\"cgd_model_10a.h5\")\n",
        "\n",
        "# RadioML2016.10b\n",
        "# model = tf.keras.models.load_model(\"cgd_model_10b.h5\")\n",
        "\n",
        "# Migou-Mod Dataset\n",
        "# model = tf.keras.models.load_model(\"CGD_MIGOU.h5\")\n",
        "\n",
        "# VUT Dataset\n",
        "# model = tf.keras.models.load_model(\"CGD_VUT.h5\")\n",
        "\n",
        "# model.summary()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "xdXxDg_xiSTV"
      },
      "source": [
        "# prediction = model.predict([X_test[:,:,:]])\n",
        "\n",
        "# Y_Pred = []; Y_Test = []; Y_Pred_SNR = []; Y_Test_SNR = []; \n",
        "# for i in range(len(prediction[:,0])):\n",
        "#     Y_Pred.append(np.argmax(prediction[i,:]))\n",
        "#     Y_Test.append(np.argmax(Y_test[i]))\n",
        "\n",
        "# Y_Pred[:20], Y_Test[:20]"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ZfPb1pzNFzJZ"
      },
      "source": [
        "\n",
        "# Visualize activation and feature map "
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "C3jyjZEgF6Rt"
      },
      "source": [
        "model_visualization(1,9000, mods, model)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "9H6_aAZ9tTiB"
      },
      "source": [
        "# Plot Confusion Matrix"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "oFxYzmHhsHbP"
      },
      "source": [
        "All_SNR_show_confusion_matrix([X_test], save=False)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "UR45Tgkh4LRQ"
      },
      "source": [
        "SNR_show_confusion_matrix(mods, lbl_SNR[:], X_test, save=False)"
      ],
      "execution_count": null,
      "outputs": []
    }
  ]
}