{"cells":[{"metadata":{},"cell_type":"markdown","source":"# References\n* JMiPOD: https://hal-utt.archives-ouvertes.fr/hal-02542075/file/J_MiPOD_vPub.pdf\n* J-UNIWARD: https://www.researchgate.net/publication/259639875_Universal_Distortion_Function_for_Steganography_in_an_Arbitrary_Domain\n* UERD: https://ieeexplore.ieee.org/abstract/document/7225122/\n\nAlso very useful, as it contains a nice introduction to the topic: http://dde.binghamton.edu/vholub/pdf/Holub_PhD_Dissertation_2014.pdf","execution_count":null},{"metadata":{},"cell_type":"markdown","source":"# Introduction\n\nSteganography is a scientific field I did not came into contact yet, everything I am writing here is the result of me trying to understand various papers about the topic. This notebook is part of this process ;)\nA consequence of this could be that most stuff I am writing here is not of interest for people which already know this field\n\nWhat I understood so far:\n* A certain payload is \"embedded\" into a cover image, in this case a JPEG image\n* The most naive way to do this embedding is for raw images using the LSB of the pixel values\n* For JPEG compressed images, not the pixel values themselves, but the DCT (discrete cosine transform) values are used for embedding\n\nThe embedding is not simply done using random pixels. Every pixel is assigned an \"embedding probability\", this probability states how likely it is that we will use this pixel for embedding. A cost function is defined and, given a certain payload, minimized with the embedding probabilities of the pixels as parameters. What is interesting is that most algorithms define their cost function in the **spatial domain**. So even when using the DCT coefficients for embedding, the cost function is not necessarily defined in this domain!\n\nMost algorithms define their cost function in such way that they are \"adaptive\". This means that it should be more likely to use a pixel for the embedding, which neighborhood shows a lot of structure. In this way it gets harder to spot the embedding.","execution_count":null},{"metadata":{},"cell_type":"markdown","source":"# Adaptive Embedding Probability\nIn the following I am trying to validate my assumption that the embedding probability is adaptive. I am using the spatial domain here, although the DCT domain is used for embedding. In http://dde.binghamton.edu/vholub/pdf/Holub_PhD_Dissertation_2014.pdf it is mentioned, that many sophisticated steganography algorithms are not necessarily detected the easiest in their embedding domain (DCT), but in the spatial domain, since they are using a cost function in the spatial domain.","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"import numpy as np\nfrom skimage import io, color\nimport matplotlib.pyplot as plt\nfrom PIL import Image\nimport seaborn as sns","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Load one example images, one cover and one example of every algorithm\nimage_cover = color.rgb2ycbcr(io.imread('/kaggle/input/alaska2-image-steganalysis/Cover/00001.jpg'))\nimage_jmipod = color.rgb2ycbcr(io.imread('/kaggle/input/alaska2-image-steganalysis/JMiPOD/00001.jpg'))\nimage_juniward = color.rgb2ycbcr(io.imread('/kaggle/input/alaska2-image-steganalysis/JUNIWARD/00001.jpg'))\nimage_uerd = color.rgb2ycbcr(io.imread('/kaggle/input/alaska2-image-steganalysis/UERD/00001.jpg'))\n\nfig, ax = plt.subplots(1,2)\nax[0].imshow(io.imread('/kaggle/input/alaska2-image-steganalysis/Cover/00001.jpg'))\nax[1].imshow(image_cover[:,:,0])\nplt.show()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"I am using only the Y channel for this experiment, the other channels should (hopefully) behave the same. \nI define structure as the intensity standard deviation within one JPEG block of 8x8 pixels.","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"# Calculate the standard deviation within the JPEG blocks (8x8)\ndef calc_block_std(image_channel):\n    \n    std_image = np.zeros((64, 64))\n    for i in range(63):\n        for j in range(63):\n            \n            std_image[i,j] = np.std(image_channel[i*8:(i+1)*8, j*8:(j+1)*8])\n                                    \n    return std_image\n\n# Calculate standard deviation for the Y channel of the cover image\ncover_y_std = calc_block_std(image_cover[:,:,0])\nplt.imshow(cover_y_std)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Now let's calculate which blocks have at least one changed pixel value with respect to the cover image","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"# Indicate which blocks have at least one changed pixel value\ndef block_changed(image_cover, image_hidden):\n    \n    changed_image = np.ones((64, 64))\n    for i in range(63):\n        for j in range(63):\n            \n            changed_image[i,j] = not np.allclose(image_cover[i*8:(i+1)*8, j*8:(j+1)*8],\n                                            image_hidden[i*8:(i+1)*8, j*8:(j+1)*8])\n                                    \n    return changed_image.astype(bool)    ","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"jmipod_diff = block_changed(image_cover[:,:,0], image_jmipod[:,:,0])\njuniward_diff = block_changed(image_cover[:,:,0], image_juniward[:,:,0])\nuerd_diff = block_changed(image_cover[:,:,0], image_uerd[:,:,0])\nfig, ax = plt.subplots(1,4)\nfig.set_figheight(15)\nfig.set_figwidth(15)\nax[0].imshow(cover_y_std)\nax[1].imshow(jmipod_diff)\nax[2].imshow(juniward_diff)\nax[3].imshow(uerd_diff)\nplt.show()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"It is clearly visible that areas which show a higher standard deviation have a higher embedding probability.","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"fig, ax = plt.subplots(1,3)\nfig.set_figheight(5)\nfig.set_figwidth(15)\n\nsns.kdeplot(cover_y_std[jmipod_diff].reshape(-1), label=\"JMiPod Changed\", ax=ax[0])\nsns.kdeplot(cover_y_std[jmipod_diff==False].reshape(-1), label=\"JMiPod Not Changed\", ax=ax[0])\nsns.kdeplot(cover_y_std[juniward_diff].reshape(-1), label=\"Juniward Changed\", ax=ax[1])\nsns.kdeplot(cover_y_std[juniward_diff==False].reshape(-1), label=\"Juniward Not Changed\", ax=ax[1])\nsns.kdeplot(cover_y_std[uerd_diff].reshape(-1), label=\"Uerd Changed\", ax=ax[2])\nsns.kdeplot(cover_y_std[uerd_diff==False].reshape(-1), label=\"Uerd Not Changed\", ax=ax[2])\nax[0].set_xlabel(\"Intensity std. within one block\")\nax[1].set_xlabel(\"Intensity std. within one block\")\nax[2].set_xlabel(\"Intensity std. within one block\")\nplt.show()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Modelling\n\nI don't have too much experience with model building, especially not for steganalysis.The core ideas of the model (extracted from multiple papers) are:\n* Make it easier for the model to converge by preprocessing the image using several high-pass filters. This is rather straight forward, the manipulation of the DCT coefficients will lead to a high-frequency \"noise\" component. Additionally we found out above that areas with a high intensity standard deviation have a high embedding probability.\n* Instead of DCT or YCbCr I am using RGB grayscale\n* We have the cover image and three different embedding algorithms --> four classes (I am neglecting here the different compression levels)\n* Instead of a dense head using Global Average Pooling, the concept of this was rather new to me","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"import tensorflow as tf\nimport os\nfrom math import ceil\nfrom skimage.io import imread\nfrom skimage import color, transform\nimport numpy as np\nfrom scipy import ndimage, misc\nfrom sklearn.model_selection import train_test_split\nimport matplotlib.pyplot as plt\nfrom tensorflow.keras.models import Sequential, Model\nfrom tensorflow.keras.layers import Dense, Conv2D, Flatten, MaxPooling2D, Activation, GlobalAveragePooling2D\nfrom tensorflow.keras.callbacks import ModelCheckpoint, EarlyStopping, ReduceLROnPlateau\nfrom tensorflow.keras.optimizers import Adam\nfrom tensorflow.keras.initializers import he_normal","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"cover_path = os.path.join(os.getcwd(), '/kaggle/input/alaska2-image-steganalysis/Cover/')\ncover_paths_labels = [(os.path.join(cover_path, f), 0) for f in os.listdir(cover_path)]\n\njmipod_path = os.path.join(os.getcwd(), '/kaggle/input/alaska2-image-steganalysis/JMiPOD/')\njmipod_paths_labels = [(os.path.join(jmipod_path, f), 1) for f in os.listdir(jmipod_path)]\n\njuniward_path = os.path.join(os.getcwd(), '/kaggle/input/alaska2-image-steganalysis/JUNIWARD/')\njuniward_paths_labels = [(os.path.join(juniward_path, f), 2) for f in os.listdir(juniward_path)]\n\nuerd_path = os.path.join(os.getcwd(), '/kaggle/input/alaska2-image-steganalysis/UERD/')\nuerd_paths_labels = [(os.path.join(uerd_path, f), 3) for f in os.listdir(uerd_path)]","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"BATCH_SIZE = 20\nSHUFFLE_BUFFER = 100\nNUM_CLASSES = 4\n\ndef _parse_function(image_path, label):\n\n    bits = tf.io.read_file(image_path)\n    image = tf.image.decode_jpeg(bits, channels=1)\n    image= tf.cast(image , tf.float32) / 255.  \n    image = tf.reshape(image, [512, 512, 1])\n    \n    label = tf.one_hot(label, NUM_CLASSES)\n    label = tf.reshape(label, (4,))\n\n    return image, label\n\ndef _augment(image, label):\n    \n    image = tf.image.random_flip_left_right(image)\n    image = tf.image.random_flip_up_down(image)\n    \n    # Rotate the image\n    k = tf.random.uniform([], minval=0, maxval=4, dtype=tf.int32)\n    image = tf.image.rot90(image, k)\n    \n    return image, label\n    \n\ndef create_dataset(image_paths_labels, augment=True, shuffle=True):\n\n    # This works with arrays as well\n    dataset = tf.data.Dataset.from_generator(lambda: image_paths_labels, (tf.string, tf.int32))\n\n    # Maps the parser on every filepath in the array. You can set the number of parallel loaders here\n    dataset = dataset.map(_parse_function, num_parallel_calls=tf.data.experimental.AUTOTUNE)\n    \n    # Set batch size\n    dataset = dataset.batch(BATCH_SIZE)\n    \n    # Augmentation\n    if augment:\n        dataset = dataset.map(_augment, num_parallel_calls=tf.data.experimental.AUTOTUNE)\n\n    # Set the number of datapoints you want to load and shuffle \n    if shuffle:\n        dataset = dataset.shuffle(SHUFFLE_BUFFER)\n    \n    dataset.prefetch(tf.data.experimental.AUTOTUNE)\n\n    return dataset","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Plot some images and compare embedding algorithms\ndata_cover = create_dataset(cover_paths_labels, augment=False, shuffle=False).take(1)\ndata_jmipod = create_dataset(jmipod_paths_labels, augment=False, shuffle=False).take(1)\ndata_juniward = create_dataset(juniward_paths_labels, augment=False, shuffle=False).take(1)\ndata_uerd = create_dataset(uerd_paths_labels, augment=False, shuffle=False).take(1)\n\nfor cov, jmi, jun, uerd in zip(data_cover, data_jmipod, data_juniward, data_uerd):\n    fig, ax = plt.subplots(2,4, figsize=(15,5))\n    ax[0, 0].imshow(np.reshape(cov[0][1], (512,512)))\n    ax[0, 1].imshow(np.reshape(jmi[0][1], (512,512)))\n    ax[0, 2].imshow(np.reshape(jun[0][1], (512,512)))\n    ax[0, 3].imshow(np.reshape(uerd[0][1], (512,512))) \n    ax[1, 0].imshow(np.abs(np.reshape(jmi[0][1] - cov[0][1], (512,512))))\n    ax[1, 1].imshow(np.abs(np.reshape(jun[0][1] - cov[0][1], (512,512))))\n    ax[1, 2].imshow(np.abs(np.reshape(uerd[0][1] - cov[0][1], (512,512))))\n    ax[1, 3].imshow(np.abs(np.reshape(uerd[0][1] - jmi[0][1], (512,512))))","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"In the following I am initializing the kernels for the first convolutional layer as high-pass filters:\n* 4 kernels of order 1. In theory there are 8 of them, but I did not include mirrored ones\n* All 4 kernels of order 2\n* 4 kernels of order 3. Same argumentation as for the kernels of order 1, potentially we would have 8. Including this order gave me a huge boost in accuracy\n* All(?) 4 3x3 edge kernels\n* 3x3 and 5x5 square kernel","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"def weight_init(shape, dtype=None):\n    \n    kernel_first_order_1 = tf.constant([[0, 0, 0, 0, 0],\n                           [0, 0, 0, 0 ,0],\n                           [0, 1, -1, 0 ,0],\n                           [0, 0, 0, 0 ,0],\n                           [0, 0, 0, 0 ,0]], dtype=tf.float32)\n    kernel_first_order_1 = tf.reshape(kernel_first_order_1, [5, 5, 1, 1])\n    \n    kernel_first_order_2 = tf.constant([[0, 0, 0, 0, 0],\n                           [0, 0, 0, 0 ,0],\n                           [0, 0, -1, 0 ,0],\n                           [0, 0, 1, 0 ,0],\n                           [0, 0, 0, 0 ,0]], dtype=tf.float32)\n    kernel_first_order_2 = tf.reshape(kernel_first_order_2, [5, 5, 1, 1])    \n    \n    kernel_first_order_3 = tf.constant([[0, 0, 0, 0, 0],\n                           [0, 0, 0, 0 ,0],\n                           [0, 0, -1, 0 ,0],\n                           [0, 0, 0, 1 ,0],\n                           [0, 0, 0, 0 ,0]], dtype=tf.float32)\n    kernel_first_order_3 = tf.reshape(kernel_first_order_3, [5, 5, 1, 1])  \n    \n    kernel_first_order_4 = tf.constant([[0, 0, 0, 0, 0],\n                           [0, 0, 0, 0 ,0],\n                           [0, 0, -1, 0 ,0],\n                           [0, 1, 0, 0 ,0],\n                           [0, 0, 0, 0 ,0]], dtype=tf.float32)\n    kernel_first_order_4 = tf.reshape(kernel_first_order_4, [5, 5, 1, 1]) \n    \n    kernel_second_order_1 = tf.constant([[0, 0, 0, 0, 0],\n                           [0, 0, 0, 0 ,0],\n                           [0, 1, -2, 1 ,0],\n                           [0, 0, 0, 0 ,0],\n                           [0, 0, 0, 0 ,0]], dtype=tf.float32)\n    kernel_second_order_1 = tf.reshape(kernel_second_order_1, [5, 5, 1, 1]) \n    \n    kernel_second_order_2 = tf.constant([[0, 0, 0, 0, 0],\n                           [0, 1, 0, 0 ,0],\n                           [0, 0, -2, 0 ,0],\n                           [0, 0, 0, 1 ,0],\n                           [0, 0, 0, 0 ,0]], dtype=tf.float32)\n    kernel_second_order_2 = tf.reshape(kernel_second_order_2, [5, 5, 1, 1]) \n    \n    kernel_second_order_3 = tf.constant([[0, 0, 0, 0, 0],\n                           [0, 0, 1, 0 ,0],\n                           [0, 0, -2, 0 ,0],\n                           [0, 0, 1, 0 ,0],\n                           [0, 0, 0, 0 ,0]], dtype=tf.float32)\n    kernel_second_order_3 = tf.reshape(kernel_second_order_3, [5, 5, 1, 1]) \n    \n    kernel_second_order_4 = tf.constant([[0, 0, 0, 0, 0],\n                           [0, 0, 0, 1 ,0],\n                           [0, 0, -2, 0 ,0],\n                           [0, 1, 0, 0 ,0],\n                           [0, 0, 0, 0 ,0]], dtype=tf.float32)\n    kernel_second_order_4 = tf.reshape(kernel_second_order_4, [5, 5, 1, 1]) \n    \n    kernel_third_order_1 = tf.constant([[0, 0, 0, 0, -1],\n                           [0, 0, 0, 3 ,0],\n                           [0, 0, -3, 0 ,0],\n                           [0, 1, 0, 0 ,0],\n                           [0, 0, 0, 0 ,0]], dtype=tf.float32)\n    kernel_third_order_1 = tf.reshape(kernel_third_order_1, [5, 5, 1, 1]) \n    \n    kernel_third_order_2 = tf.constant([[0, 0, 0, 0, 0],\n                           [0, 1, 0, 0 ,0],\n                           [0, 0, -3, 0 ,0],\n                           [0, 0, 0, 3 ,0],\n                           [0, 0, 0, 0 ,-1]], dtype=tf.float32)\n    kernel_third_order_2 = tf.reshape(kernel_third_order_2, [5, 5, 1, 1]) \n    \n    kernel_third_order_3 = tf.constant([[0, 0, 0, 0, 0],\n                           [0, 0, 0, 0 ,0],\n                           [-1, 3, -3, 1 ,0],\n                           [0, 0, 0, 0 ,0],\n                           [0, 0, 0, 0 ,0]], dtype=tf.float32)\n    kernel_third_order_3 = tf.reshape(kernel_third_order_3, [5, 5, 1, 1]) \n    \n    kernel_third_order_4 = tf.constant([[0, 0, 0, 0, 0],\n                           [0, 0, 1, 0 ,0],\n                           [0, 0, -3, 0 ,0],\n                           [0, 0, 3, 0 ,0],\n                           [0, 0, -1, 0 ,0]], dtype=tf.float32)\n    kernel_third_order_4 = tf.reshape(kernel_third_order_4, [5, 5, 1, 1]) \n    \n    kernel_edge_three_1 = tf.constant([[0, 0, 0, 0, 0],\n                           [0, -1, 2, -1 ,0],\n                           [0, 2, -4, 2 ,0],\n                           [0, 0, 0, 0 ,0],\n                           [0, 0, 0, 0 ,0]], dtype=tf.float32)\n    kernel_edge_three_1 = tf.reshape(kernel_edge_three_1, [5, 5, 1, 1])\n    \n    kernel_edge_three_2 = tf.constant([[0, 0, 0, 0, 0],\n                           [0, 0, 0, 0 ,0],\n                           [0, 2, -4, 2 ,0],\n                           [0, -1, 2, -1 ,0],\n                           [0, 0, 0, 0 ,0]], dtype=tf.float32)\n    kernel_edge_three_2 = tf.reshape(kernel_edge_three_2, [5, 5, 1, 1]) \n    \n    kernel_edge_three_3 = tf.constant([[0, 0, 0, 0, 0],\n                           [0, -1, 2, 0 ,0],\n                           [0, 2, -4, 0 ,0],\n                           [0, -1, 2, 0 ,0],\n                           [0, 0, 0, 0 ,0]], dtype=tf.float32)\n    kernel_edge_three_3 = tf.reshape(kernel_edge_three_3, [5, 5, 1, 1]) \n    \n    kernel_edge_three_4 = tf.constant([[0, 0, 0, 0, 0],\n                           [0, 0, 2, -1 ,0],\n                           [0, 0, -4, 2 ,0],\n                           [0, 0, 2, -1 ,0],\n                           [0, 0, 0, 0 ,0]], dtype=tf.float32)\n    kernel_edge_three_4 = tf.reshape(kernel_edge_three_4, [5, 5, 1, 1]) \n    \n    square_kernel_1 = tf.constant([[-1, 2, -2, 2, -1],\n                           [2, -6, 8, -6 ,2],\n                           [-2, 8, -12, 8 ,-2],\n                           [2, -6, 8, -6 ,2],\n                           [-1, 2, -2, 2 ,-1]], dtype=tf.float32)\n    square_kernel_1 = tf.reshape(square_kernel_1, [5, 5, 1, 1]) \n    \n    \n    square_kernel_2 = tf.constant([[0, 0, 0, 0, 0],\n                           [0, -1, 2, -1 ,0],\n                           [0, 2, -4, 2 ,0],\n                           [0, -1, 2, -1 ,0],\n                           [0, 0, 0, 0 ,0]], dtype=tf.float32)\n    square_kernel_2 = tf.reshape(square_kernel_2, [5, 5, 1, 1]) \n    \n    kernel_sum = tf.stack([kernel_first_order_1, \n                           kernel_first_order_2,\n                           kernel_first_order_3,\n                           kernel_first_order_4,\n                           kernel_second_order_1,\n                           kernel_second_order_2,\n                           kernel_second_order_3,\n                           kernel_second_order_4,\n                           kernel_third_order_1,\n                           kernel_third_order_2,\n                           kernel_third_order_3,\n                           kernel_third_order_4,\n                           kernel_edge_three_1,\n                           kernel_edge_three_2,\n                           kernel_edge_three_3,\n                           kernel_edge_three_4,\n                           square_kernel_1,\n                           square_kernel_2], axis=3)\n    kernel_collection = tf.reshape(kernel_sum, [5, 5, 1, 18])\n    \n    return kernel_collection","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Create training/validation dataset\npaths_labels = cover_paths_labels + jmipod_paths_labels + juniward_paths_labels + uerd_paths_labels\npaths_labels_train, paths_labels_valid = train_test_split(paths_labels, shuffle=True, test_size=0.20)\n\ndata_train = create_dataset(paths_labels_train)\ndata_valid = create_dataset(paths_labels_valid)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"model_dummy = Sequential()\nmodel_dummy.add(Conv2D(18, kernel_initializer=weight_init, kernel_size=5, padding=\"valid\", input_shape=(512,512, 1)))\n\n# Some testing\ndata_dummy = create_dataset(paths_labels_train).take(1)\ndummy_images = model_dummy.predict(data_dummy)\n\nfig, ax = plt.subplots(1,2, figsize=(15,5))\nax[0].hist(np.reshape(dummy_images[0, :,:, 13]*255, (508*508,)))\nax[1].imshow(dummy_images[0, :,:, 13])","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Okay we don't see too much in these images, let's try to take the absolute value. This method has been mentioned in a few papers, others don't use that, but rather use a parametrized ReLU version to handle negative values.","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"from tensorflow.keras import layers\n\nclass AbsLayer(layers.Layer):\n    \n    def get_config(self):\n        base_config = super(AbsLayer, self).get_config()\n        return dict(list(base_config.items()))\n    \n    def __init__(self, **kwargs):\n        super(AbsLayer, self).__init__(**kwargs)\n        \n    def build(self, input_shape):\n        super(AbsLayer, self).build(input_shape)\n    \n    def call(self, inputs):\n        return tf.abs(inputs)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"model_dummy_abs = Sequential()\nmodel_dummy_abs.add(Conv2D(18, kernel_initializer=weight_init, kernel_size=5, padding=\"valid\", input_shape=(512,512, 1)))\nmodel_dummy_abs.add(AbsLayer())\n\ndummy_images_abs = model_dummy_abs.predict(data_dummy)\n\nfig, ax = plt.subplots(1,2, figsize=(15,5))\nax[0].hist(np.reshape(dummy_images_abs[0, :,:, 1]*255, (508*508,)))\nax[1].imshow(dummy_images_abs[0, :,:, 1])","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"This looks much better, but we still have a problem due to the large dynamic range of the pixel values. Let's truncate the pixel values.\nWhen using truncation we should not use max. pooling! I learned that the hard way (kind of makes sense).","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"# Define a custom activation function which truncates the pixel values\ndef act_trunc(inputs):\n    TRUNCATION_VAL = tf.constant(10./255., dtype=tf.float32)\n    return tf.clip_by_value(inputs, -TRUNCATION_VAL, TRUNCATION_VAL)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"model_dummy_trunc = Sequential()\nmodel_dummy_trunc.add(Conv2D(18, activation=act_trunc, kernel_initializer=weight_init, kernel_size=5, padding=\"valid\", input_shape=(512,512, 1)))\nmodel_dummy_trunc.add(AbsLayer())\n\ndummy_images_trunc = model_dummy_trunc.predict(data_dummy)\n\nfig, ax = plt.subplots(1,2, figsize=(15,5))\nax[0].hist(np.reshape(dummy_images_trunc[0, :,:, 1]*255, (508*508,)))\nax[1].imshow(dummy_images_trunc[0, :,:, 1])","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"reduce_lr = ReduceLROnPlateau(monitor='val_loss', factor=0.2,\n                              patience=2, min_lr=1e-7, verbose=1)\n\n# Build the model. I am not using truncation here, I could not get it to work reliably.\nmodel = Sequential()\nmodel.add(Conv2D(18, strides=(2,2), kernel_initializer=weight_init, kernel_size=5, padding=\"valid\", input_shape=(512,512, 1)))\nmodel.add(AbsLayer())\nmodel.add(Conv2D(32, kernel_size=2, strides=(2,2), padding=\"same\", kernel_initializer=he_normal()))\nmodel.add(Activation('relu'))\nmodel.add(Conv2D(64, kernel_size=2, strides=(2,2), padding=\"same\", kernel_initializer=he_normal()))\nmodel.add(Activation('relu'))\nmodel.add(Conv2D(64, kernel_size=2, padding=\"same\", kernel_initializer=he_normal()))\nmodel.add(Activation('relu'))\nmodel.add(Conv2D(128, kernel_size=2, strides=(2,2), padding=\"same\", kernel_initializer=he_normal()))\nmodel.add(Activation('relu'))\nmodel.add(Conv2D(256, kernel_size=2, strides=(2,2), padding=\"same\", kernel_initializer=he_normal()))\nmodel.add(Activation('relu'))\nmodel.add(Conv2D(512, kernel_size=2, strides=(2,2), padding=\"same\", kernel_initializer=he_normal()))\nmodel.add(Activation('relu'))\nmodel.add(Conv2D(1024, kernel_size=2, strides=(2,2), padding=\"same\", kernel_initializer=he_normal()))\nmodel.add(Activation('relu'))\nmodel.add(GlobalAveragePooling2D())\nmodel.add(Dense(4,  activation='softmax'))\nmodel.compile(optimizer=Adam(lr=0.0001), loss='categorical_crossentropy', metrics=['accuracy'])\n\nmodel.summary()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Just train for a few steps, offline I achieved roughly ~86% with a similar model","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"model.fit(x=data_train, validation_data=data_valid,\n          steps_per_epoch=1000, validation_steps=10, callbacks=[reduce_lr])","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Inference","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"test_path = os.path.join(os.getcwd(), '/kaggle/input/alaska2-image-steganalysis/Test/')\ntest_labels = [(os.path.join(test_path, f), -1) for f in os.listdir(test_path)]\n\ndata_test = create_dataset(test_labels, shuffle=False, augment=False)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"from tensorflow.keras.models import load_model\n\ncustom_obj = {}\ncustom_obj['weight_init'] = weight_init\ncustom_obj['AbsLayer'] = AbsLayer\n\npredictions = model.predict(data_test, verbose=1)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# We are not interested in the specific embedding class but rather whether it is a cover or stego image\nprediction_stego = [np.sum(pred[1:4]) for pred in predictions]\n\nimport pandas as pd\ntest_path = os.path.join(os.getcwd(), '/kaggle/input/alaska2-image-steganalysis/Test/')\nsub = pd.DataFrame({'Id': os.listdir(test_path),\n                    'Label': prediction_stego})\n\nsub.to_csv(\"submission.csv\", index=False)","execution_count":null,"outputs":[]}],"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat":4,"nbformat_minor":4}