QDL-CMFD.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "source": [
    "## QDL-CMFD\r\n",
    "### a Quality-independent and Deep Learning-based Copy-Move image Forgery Detection method.\r\n",
    "#### _ Mehrad Aria"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "import tensorflow as tf\r\n",
    "from tensorflow.keras.layers import Conv2D, MaxPooling2D\r\n",
    "from tensorflow.keras.layers import Layer, Input, Lambda\r\n",
    "from tensorflow.keras.layers import BatchNormalization, Activation, Concatenate\r\n",
    "from tensorflow.keras import regularizers\r\n",
    "from tensorflow.keras.models import Model\r\n",
    "from tensorflow.keras.applications.vgg16 import preprocess_input\r\n",
    "from tensorflow.keras import backend as K\r\n",
    "\r\n",
    "import numpy as np\r\n",
    "import collections\r\n",
    "from itertools import chain\r\n",
    "import pickle\r\n",
    "\r\n",
    "from libsvm import svmutil\r\n",
    "import scipy.signal as signal\r\n",
    "import scipy.special as special\r\n",
    "import scipy.optimize as optimize\r\n",
    "\r\n",
    "from PIL import Image\r\n",
    "import skimage.io\r\n",
    "import skimage.transform\r\n",
    "import cv2\r\n",
    "\r\n",
    "import matplotlib.pyplot as plt\r\n",
    "from __future__ import print_function\r\n",
    "import warnings\r\n",
    "warnings.filterwarnings(\"ignore\")\r\n",
    "%matplotlib inline\r\n",
    "%load_ext autoreload\r\n",
    "%autoreload 2"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "def normalize_kernel(kernel):\r\n",
    "    return kernel / np.sum(kernel)\r\n",
    "\r\n",
    "def gaussian_kernel2d(n, sigma):\r\n",
    "    Y, X = np.indices((n, n)) - int(n/2)\r\n",
    "    gaussian_kernel = 1 / (2 * np.pi * sigma ** 2) * np.exp(-(X ** 2 + Y ** 2) / (2 * sigma ** 2))\r\n",
    "    return normalize_kernel(gaussian_kernel)\r\n",
    "\r\n",
    "def local_mean(image, kernel):\r\n",
    "    return signal.convolve2d(image, kernel, 'same')"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "def local_deviation(image, local_mean, kernel):\r\n",
    "    sigma = image ** 2\r\n",
    "    sigma = signal.convolve2d(sigma, kernel, 'same')\r\n",
    "    return np.sqrt(np.abs(local_mean ** 2 - sigma))"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "def calculate_mscn_coefficients(image, kernel_size=6, sigma=7/6):\r\n",
    "    C = 1/255\r\n",
    "    kernel = gaussian_kernel2d(kernel_size, sigma=sigma)\r\n",
    "    local_mean = signal.convolve2d(image, kernel, 'same')\r\n",
    "    local_var = local_deviation(image, local_mean, kernel)\r\n",
    "\r\n",
    "    return (image - local_mean) / (local_var + C)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "def generalized_gaussian_dist(x, alpha, sigma):\r\n",
    "    beta = sigma * np.sqrt(special.gamma(1 / alpha) / special.gamma(3 / alpha))\r\n",
    "\r\n",
    "    coefficient = alpha / (2 * beta() * special.gamma(1 / alpha))\r\n",
    "    return coefficient * np.exp(-(np.abs(x) / beta) ** alpha)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "def calculate_pair_product_coefficients(mscn_coefficients):\r\n",
    "    return collections.OrderedDict({\r\n",
    "        'mscn': mscn_coefficients,\r\n",
    "        'horizontal': mscn_coefficients[:, :-1] * mscn_coefficients[:, 1:],\r\n",
    "        'vertical': mscn_coefficients[:-1, :] * mscn_coefficients[1:, :],\r\n",
    "        'main_diagonal': mscn_coefficients[:-1, :-1] * mscn_coefficients[1:, 1:],\r\n",
    "        'secondary_diagonal': mscn_coefficients[1:, :-1] * mscn_coefficients[:-1, 1:]\r\n",
    "    })"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "def asymmetric_generalized_gaussian(x, nu, sigma_l, sigma_r):\r\n",
    "    def beta(sigma):\r\n",
    "        return sigma * np.sqrt(special.gamma(1 / nu) / special.gamma(3 / nu))\r\n",
    "\r\n",
    "    coefficient = nu / ((beta(sigma_l) + beta(sigma_r)) * special.gamma(1 / nu))\r\n",
    "    f = lambda x, sigma: coefficient * np.exp(-(x / beta(sigma)) ** nu)\r\n",
    "\r\n",
    "    return np.where(x < 0, f(-x, sigma_l), f(x, sigma_r))"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "def asymmetric_generalized_gaussian_fit(x):\r\n",
    "    def estimate_phi(alpha):\r\n",
    "        numerator = special.gamma(2 / alpha) ** 2\r\n",
    "        denominator = special.gamma(1 / alpha) * special.gamma(3 / alpha)\r\n",
    "        return numerator / denominator\r\n",
    "\r\n",
    "    def estimate_r_hat(x):\r\n",
    "        size = np.prod(x.shape)\r\n",
    "        return (np.sum(np.abs(x)) / size) ** 2 / (np.sum(x ** 2) / size)\r\n",
    "\r\n",
    "    def estimate_R_hat(r_hat, gamma):\r\n",
    "        numerator = (gamma ** 3 + 1) * (gamma + 1)\r\n",
    "        denominator = (gamma ** 2 + 1) ** 2\r\n",
    "        return r_hat * numerator / denominator\r\n",
    "\r\n",
    "    def mean_squares_sum(x, filter = lambda z: z == z):\r\n",
    "        filtered_values = x[filter(x)]\r\n",
    "        squares_sum = np.sum(filtered_values ** 2)\r\n",
    "        return squares_sum / ((filtered_values.shape))\r\n",
    "\r\n",
    "    def estimate_gamma(x):\r\n",
    "        left_squares = mean_squares_sum(x, lambda z: z < 0)\r\n",
    "        right_squares = mean_squares_sum(x, lambda z: z >= 0)\r\n",
    "\r\n",
    "        return np.sqrt(left_squares) / np.sqrt(right_squares)\r\n",
    "\r\n",
    "    def estimate_alpha(x):\r\n",
    "        r_hat = estimate_r_hat(x)\r\n",
    "        gamma = estimate_gamma(x)\r\n",
    "        R_hat = estimate_R_hat(r_hat, gamma)\r\n",
    "\r\n",
    "        solution = optimize.root(lambda z: estimate_phi(z) - R_hat, [0.2]).x\r\n",
    "\r\n",
    "        return solution[0]\r\n",
    "\r\n",
    "    def estimate_sigma(x, alpha, filter = lambda z: z < 0):\r\n",
    "        return np.sqrt(mean_squares_sum(x, filter))\r\n",
    "\r\n",
    "    def estimate_mean(alpha, sigma_l, sigma_r):\r\n",
    "        return (sigma_r - sigma_l) * constant * (special.gamma(2 / alpha) / special.gamma(1 / alpha))\r\n",
    "\r\n",
    "    alpha = estimate_alpha(x)\r\n",
    "    sigma_l = estimate_sigma(x, alpha, lambda z: z < 0)\r\n",
    "    sigma_r = estimate_sigma(x, alpha, lambda z: z >= 0)\r\n",
    "\r\n",
    "    constant = np.sqrt(special.gamma(1 / alpha) / special.gamma(3 / alpha))\r\n",
    "    mean = estimate_mean(alpha, sigma_l, sigma_r)\r\n",
    "\r\n",
    "    return alpha, mean, sigma_l, sigma_r"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "def calculate_brisque_features(image, kernel_size=7, sigma=7/6):\r\n",
    "    def calculate_features(coefficients_name, coefficients, accum=np.array([])):\r\n",
    "        alpha, mean, sigma_l, sigma_r = asymmetric_generalized_gaussian_fit(coefficients)\r\n",
    "\r\n",
    "        if coefficients_name == 'mscn':\r\n",
    "            var = (sigma_l ** 2 + sigma_r ** 2) / 2\r\n",
    "            return [alpha, var]\r\n",
    "\r\n",
    "        return [alpha, mean, sigma_l ** 2, sigma_r ** 2]\r\n",
    "\r\n",
    "    mscn_coefficients = calculate_mscn_coefficients(image, kernel_size, sigma)\r\n",
    "    coefficients = calculate_pair_product_coefficients(mscn_coefficients)\r\n",
    "\r\n",
    "    features = [calculate_features(name, coeff) for name, coeff in coefficients.items()]\r\n",
    "    flatten_features = list(chain.from_iterable(features))\r\n",
    "    return np.array(flatten_features)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "def plot_histogram(x, label):\r\n",
    "    n, bins = np.histogram(x.ravel(), bins=50)\r\n",
    "    n = n / np.max(n)\r\n",
    "    plt.plot(bins[:-1], n, label=label, marker='o')"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "gray_image = skimage.color.rgb2gray(skimage.io.imread('./Images/LR.jpg'))\r\n",
    "_ = skimage.io.imshow('./Images/LR.jpg')"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "mscn_coefficients = calculate_mscn_coefficients(gray_image, 7, 7/6)\r\n",
    "coefficients = calculate_pair_product_coefficients(mscn_coefficients)\r\n",
    "\r\n",
    "%matplotlib inline\r\n",
    "plt.rcParams[\"figure.figsize\"] = 12, 11\r\n",
    "\r\n",
    "for name, coeff in coefficients.items():\r\n",
    "    plot_histogram(coeff.ravel(), name)\r\n",
    "\r\n",
    "plt.axis([-2.5, 2.5, 0, 1.05])\r\n",
    "plt.legend(prop={'size': 16})\r\n",
    "plt.show()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "brisque_features = calculate_brisque_features(gray_image, kernel_size=7, sigma=7/6)\r\n",
    "\r\n",
    "downscaled_image = cv2.resize(gray_image, None, fx=1/2, fy=1/2, interpolation = cv2.INTER_CUBIC)\r\n",
    "downscale_brisque_features = calculate_brisque_features(downscaled_image, kernel_size=7, sigma=7/6)\r\n",
    "\r\n",
    "brisque_features = np.concatenate((brisque_features, downscale_brisque_features))\r\n",
    "\r\n",
    "def scale_features(features):\r\n",
    "    with open('./BRISQUE/normalize.pickle', 'rb') as handle:\r\n",
    "        scale_params = pickle.load(handle)\r\n",
    "\r\n",
    "    min_ = np.array(scale_params['min_'])\r\n",
    "    max_ = np.array(scale_params['max_'])\r\n",
    "\r\n",
    "    return -1 + (2.0 / (max_ - min_) * (features - min_))\r\n",
    "\r\n",
    "def calculate_image_quality_score(brisque_features):\r\n",
    "    model = svmutil.svm_load_model('./BRISQUE/brisque_svm.txt')\r\n",
    "    scaled_brisque_features = scale_features(brisque_features)\r\n",
    "\r\n",
    "    x, idx = svmutil.gen_svm_nodearray(\r\n",
    "        scaled_brisque_features,\r\n",
    "        isKernel=(model.param.kernel_type == svmutil.PRECOMPUTED))\r\n",
    "\r\n",
    "    nr_classifier = 1\r\n",
    "    prob_estimates = (svmutil.c_double * nr_classifier)()\r\n",
    "    result = svmutil.libsvm.svm_predict_probability(model, x, prob_estimates)\r\n",
    "    Nresult = 100 - result\r\n",
    "    return Nresult\r\n",
    "\r\n",
    "calculate_image_quality_score(brisque_features)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "! python ./OSRGAN/OSRGAN.py --model_path OSRGAN/Model/OSRGAN.pth --input Images/LR.jpg"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "gray_image = skimage.color.rgb2gray(skimage.io.imread('./Images/SR.jpg'))\r\n",
    "_ = skimage.io.imshow('./Images/SR.jpg')\r\n",
    "\r\n",
    "mscn_coefficients = calculate_mscn_coefficients(gray_image, 7, 7/6)\r\n",
    "coefficients = calculate_pair_product_coefficients(mscn_coefficients)\r\n",
    "\r\n",
    "%matplotlib inline\r\n",
    "plt.rcParams[\"figure.figsize\"] = 12, 11\r\n",
    "\r\n",
    "for name, coeff in coefficients.items():\r\n",
    "    plot_histogram(coeff.ravel(), name)\r\n",
    "\r\n",
    "plt.axis([-2.5, 2.5, 0, 1.05])\r\n",
    "plt.legend(prop={'size': 16})\r\n",
    "plt.show()\r\n",
    "\r\n",
    "brisque_features = calculate_brisque_features(gray_image, kernel_size=7, sigma=7/6)\r\n",
    "\r\n",
    "downscaled_image = cv2.resize(gray_image, None, fx=1/2, fy=1/2, interpolation = cv2.INTER_CUBIC)\r\n",
    "downscale_brisque_features = calculate_brisque_features(downscaled_image, kernel_size=7, sigma=7/6)\r\n",
    "\r\n",
    "brisque_features = np.concatenate((brisque_features, downscale_brisque_features))\r\n",
    "\r\n",
    "calculate_image_quality_score(brisque_features)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "def std_norm_along_chs(x) :\r\n",
    "    avg = K.mean(x, axis=-1, keepdims=True)\r\n",
    "    std = K.maximum(1e-4, K.std(x, axis=-1, keepdims=True))\r\n",
    "    return (x - avg) / std\r\n",
    "\r\n",
    "def BnInception(x, nb_inc=16, inc_filt_list=[(1,1), (3,3), (5,5)], name='uinc') :\r\n",
    "    uc_list = []\r\n",
    "    for idx, ftuple in enumerate( inc_filt_list ) :\r\n",
    "        uc = Conv2D( nb_inc, ftuple, activation='linear', padding='same', name=name+'_c%d' % idx)(x)\r\n",
    "        uc_list.append(uc)\r\n",
    "    if ( len( uc_list ) > 1 ) :\r\n",
    "        uc_merge = Concatenate( axis=-1, name=name+'_merge')(uc_list)\r\n",
    "    else :\r\n",
    "        uc_merge = uc_list[0]\r\n",
    "    uc_norm = BatchNormalization(name=name+'_bn')(uc_merge)\r\n",
    "    xn = Activation('relu', name=name+'_re')(uc_norm)\r\n",
    "    return xn\r\n",
    "\r\n",
    "class SelfCorrelationPercPooling( Layer ) :\r\n",
    "    def __init__( self, nb_pools=256, **kwargs ) :\r\n",
    "        self.nb_pools = nb_pools\r\n",
    "        super( SelfCorrelationPercPooling, self ).__init__( **kwargs )\r\n",
    "    def build( self, input_shape ) :\r\n",
    "        self.built = True\r\n",
    "    def call( self, x, mask=None ) :\r\n",
    "        bsize, nb_rows, nb_cols, nb_feats = K.int_shape( x )\r\n",
    "        nb_maps = nb_rows * nb_cols\r\n",
    "        x_3d = K.reshape( x, tf.stack( [ -1, nb_maps, nb_feats ] ) )\r\n",
    "        x_corr_3d = tf.matmul( x_3d, x_3d, transpose_a = False, transpose_b = True ) / nb_feats\r\n",
    "        x_corr = K.reshape( x_corr_3d, tf.stack( [ -1, nb_rows, nb_cols, nb_maps ] ) )\r\n",
    "        if ( self.nb_pools is not None ) :\r\n",
    "            ranks = K.cast( K.round( tf.linspace( 1., nb_maps - 1, self.nb_pools ) ), 'int32' )\r\n",
    "        else :\r\n",
    "            ranks = tf.range( 1, nb_maps, dtype = 'int32' )\r\n",
    "        x_sort, _ = tf.nn.top_k( x_corr, k = nb_maps, sorted = True )\r\n",
    "        x_f1st_sort = K.permute_dimensions( x_sort, ( 3, 0, 1, 2 ) )\r\n",
    "        x_f1st_pool = tf.gather( x_f1st_sort, ranks )\r\n",
    "        x_pool = K.permute_dimensions( x_f1st_pool, ( 1, 2, 3, 0 ) )\r\n",
    "        return x_pool\r\n",
    "    def compute_output_shape( self, input_shape ) :\r\n",
    "        bsize, nb_rows, nb_cols, nb_feats = input_shape\r\n",
    "        nb_pools = self.nb_pools if ( self.nb_pools is not None ) else ( nb_rows * nb_cols - 1 )\r\n",
    "        return tuple( [ bsize, nb_rows, nb_cols, nb_pools ] )\r\n",
    "\r\n",
    "class BilinearUpSampling2D( Layer ) :\r\n",
    "    def call( self, x, mask=None ) :\r\n",
    "        bsize, nb_rows, nb_cols, nb_filts = K.int_shape(x)\r\n",
    "        new_size = tf.constant( [ nb_rows * 2, nb_cols * 2 ], dtype = tf.int32 )\r\n",
    "        return tf.image.resize( x, new_size)\r\n",
    "    def compute_output_shape( self, input_shape ) :\r\n",
    "        bsize, nb_rows, nb_cols, nb_filts = input_shape\r\n",
    "        return tuple( [ bsize, nb_rows * 2, nb_cols * 2, nb_filts ] )\r\n",
    "\r\n",
    "class ResizeBack( Layer ) :\r\n",
    "    def call( self, x ) :\r\n",
    "        t, r = x\r\n",
    "        new_size = [ tf.shape(input=r)[1], tf.shape(input=r)[2] ]\r\n",
    "        return tf.image.resize( t, new_size)\r\n",
    "    def compute_output_shape( self, input_shapes ) :\r\n",
    "        tshape, rshape = input_shapes\r\n",
    "        return ( tshape[0], ) + rshape[1:3] + ( tshape[-1], )\r\n",
    "    \r\n",
    "class Preprocess( Layer ) :\r\n",
    "\r\n",
    "    def call( self, x, mask=None ) :\r\n",
    "        bsize, nb_rows, nb_cols, nb_colors = K.int_shape(x)\r\n",
    "        if (nb_rows != 256) or (nb_cols !=256) :\r\n",
    "            x256 = tf.image.resize( x,\r\n",
    "                                             [256, 256],\r\n",
    "                                             method=tf.image.ResizeMethod.BILINEAR,\r\n",
    "                                             name='resize' )\r\n",
    "        else :\r\n",
    "            x256 = x\r\n",
    "        if K.dtype(x) == 'float32' :\r\n",
    "            xout = x256\r\n",
    "        else :\r\n",
    "            xout = preprocess_input( x256 )\r\n",
    "        return xout\r\n",
    "    def compute_output_shape( self, input_shape ) :\r\n",
    "        return (input_shape[0], 256, 256, 3)\r\n",
    "\r\n",
    "def create_qdlcmfd_similarity_subnetwork( img_shape=(256,256,3),\r\n",
    "                                   nb_pools=100,\r\n",
    "                                   name='QDL-CMFD-S' ) :\r\n",
    "    img_input = Input( shape=img_shape, name=name+'_in' )\r\n",
    "    bname = name + '_cnn'\r\n",
    "\r\n",
    "    x = Conv2D(64, (3, 3), activation=tf.keras.layers.LeakyReLU(), kernel_regularizer=regularizers.L1L2(l2=1e-4), padding='same', name=bname+'_c1')(img_input)\r\n",
    "    x = BatchNormalization(axis = -1, name=bname+'_b1')(x)\r\n",
    "    x = Conv2D(256, (3, 3), activation=tf.keras.layers.LeakyReLU(), kernel_regularizer=regularizers.L1L2(l2=1e-4), padding='same', name=bname+'_c2')(x)\r\n",
    "    x = BatchNormalization(axis = -1, name=bname+'_b2')(x)\r\n",
    "    x = Conv2D(512, (3, 3), activation=tf.keras.layers.LeakyReLU(), kernel_regularizer=regularizers.L1L2(l2=1e-4), padding='same', name=bname+'_c3')(x)\r\n",
    "    x = BatchNormalization(axis = -1, name=bname+'_b3')(x)\r\n",
    "    x = MaxPooling2D((2, 2), strides=(2, 2), name=bname+'_mp')(x)\r\n",
    "    x = Activation(std_norm_along_chs, name=bname+'_sn')(x)\r\n",
    "\r\n",
    "    bname = name + '_corr'\r\n",
    "    xcorr = SelfCorrelationPercPooling(name=bname+'_corr')(x)\r\n",
    "    xn = BatchNormalization(name=bname+'_bn')(xcorr)\r\n",
    "    patch_list = [(1,1),(3,3),(5,5)]\r\n",
    "\r\n",
    "    bname = name + '_dconv'\r\n",
    "    f16  = BnInception( xn, 8, patch_list, name =bname+'_mpf')\r\n",
    "    f32  = BilinearUpSampling2D( name=bname+'_bx2')( f16 )\r\n",
    "    dx32 = BnInception( f32, 6, patch_list, name=bname+'_dx2')\r\n",
    "    f64a = BilinearUpSampling2D( name=bname+'_bx4a')( f32 )\r\n",
    "    f64b = BilinearUpSampling2D( name=bname+'_bx4b')( dx32 )\r\n",
    "    f64  = Concatenate(axis=-1, name=name+'_dx4_m')([f64a, f64b])\r\n",
    "    dx64 = BnInception( f64, 4, patch_list, name=bname+'_dx4')\r\n",
    "    f128a = BilinearUpSampling2D( name=bname+'_bx8a')( f64a )\r\n",
    "    f128b = BilinearUpSampling2D( name=bname+'_bx8b')( dx64 )\r\n",
    "    f128  = Concatenate(axis=-1, name=name+'_dx8_m')([f128a, f128b])\r\n",
    "    dx128 = BnInception( f128, 2, patch_list, name=bname+'_dx8')\r\n",
    "    f256a = BilinearUpSampling2D( name=bname+'_bx16a')( f128a )\r\n",
    "    f256b = BilinearUpSampling2D( name=bname+'_bx16b')( dx128 )\r\n",
    "    f256  = Concatenate(axis=-1, name=name+'_dx16_m')([f256a,f256b])\r\n",
    "    dx256 = BnInception( f256, 2, patch_list, name=bname+'_dx16')\r\n",
    "    fm256 = Concatenate(axis=-1,name=name+'_mfeat')([f256a,dx256])\r\n",
    "    masks = BnInception( fm256, 2, [(5,5),(7,7),(11,11)], name=bname+'_dxF')\r\n",
    "\r\n",
    "    pred_mask = Conv2D(1, (3,3), activation='sigmoid', name=name+'_pred_mask', padding='same')(masks)\r\n",
    "\r\n",
    "    model = Model(inputs=img_input, outputs=pred_mask, name=name)\r\n",
    "    return model\r\n",
    "\r\n",
    "def create_qdlcmfd_manipulation_subnetwork( img_shape=(256,256,3),\r\n",
    "                                     name='QDL-CMFD-M' ) :\r\n",
    "    img_input = Input( shape = img_shape, name = name+'_in' )\r\n",
    "    bname = name + '_cnn'\r\n",
    "\r\n",
    "    x = Conv2D(64, (3, 3), activation=tf.keras.layers.LeakyReLU(), kernel_regularizer=regularizers.L1L2(l2=1e-4), padding='same', name=bname+'_c1')(img_input)\r\n",
    "    x = BatchNormalization(axis = -1, name=bname+'_b1')(x)\r\n",
    "    x = Conv2D(256, (3, 3), activation=tf.keras.layers.LeakyReLU(), kernel_regularizer=regularizers.L1L2(l2=1e-4), padding='same', name=bname+'_c2')(x)\r\n",
    "    x = BatchNormalization(axis = -1, name=bname+'_b2')(x)\r\n",
    "    x = Conv2D(512, (3, 3), activation=tf.keras.layers.LeakyReLU(), kernel_regularizer=regularizers.L1L2(l2=1e-4), padding='same', name=bname+'_c3')(x)\r\n",
    "    x = BatchNormalization(axis = -1, name=bname+'_b3')(x)\r\n",
    "    x = MaxPooling2D((2, 2), strides=(2, 2), name=bname+'_mp')(x)\r\n",
    "\r\n",
    "    patch_list = [(1,1),(3,3),(5,5)]\r\n",
    "    bname = name + '_dconv'\r\n",
    "    f16 = BnInception( x, 8, patch_list, name =bname+'_mpf')\r\n",
    "    f32  = BilinearUpSampling2D(name=bname+'_bx2')( f16 )\r\n",
    "    dx32 = BnInception( f32, 6, patch_list, name=bname+'_dx2')\r\n",
    "    f64  = BilinearUpSampling2D(name=bname+'_bx4')( dx32 )\r\n",
    "    dx64 = BnInception( f64, 4, patch_list, name=bname+'_dx4')\r\n",
    "    f128  = BilinearUpSampling2D(name=bname+'_bx8')( dx64 )\r\n",
    "    dx128 = BnInception( f128, 2, patch_list, name=bname+'_dx8')\r\n",
    "    f256  = BilinearUpSampling2D(name=bname+'_bx16')( dx128 )\r\n",
    "    dx256 = BnInception( f256, 2, [(5,5),(7,7),(11,11)], name=bname+'_dx16')\r\n",
    "\r\n",
    "    pred_mask = Conv2D(1, (3,3), activation='sigmoid', name=bname+'_pred_mask', padding='same')(dx256)\r\n",
    "\r\n",
    "    model = Model(inputs=img_input, outputs=pred_mask, name = bname)\r\n",
    "    return model\r\n",
    "\r\n",
    "def create_QDLCMFD_test_model( weight_file=None ) :\r\n",
    "    similarity_subnetwork = create_qdlcmfd_similarity_subnetwork()\r\n",
    "    manipulation_subnetwork = create_qdlcmfd_manipulation_subnetwork()\r\n",
    "    QDLCMFD_S = Model( inputs=similarity_subnetwork.inputs, outputs=similarity_subnetwork.layers[-2].output)\r\n",
    "    QDLCMFD_M = Model( inputs=manipulation_subnetwork.inputs, outputs=manipulation_subnetwork.layers[-2].output)\r\n",
    "\r\n",
    "    img_raw = Input( shape=(None,None,3), name='image_in')\r\n",
    "    img_in = Preprocess( name='preprocess')( img_raw )\r\n",
    "    subnetwork1 = QDLCMFD_S( img_in )\r\n",
    "    subnetwork2 = QDLCMFD_M( img_in )\r\n",
    "    merged_network = Concatenate(axis=-1, name='merge')([subnetwork1, subnetwork2])\r\n",
    "    f = BnInception( merged_network, 3, name='fusion' )\r\n",
    "    mask_out = Conv2D( 3, (3,3), padding='same', activation='softmax', name='pred_mask')(f)\r\n",
    "    mask_out = ResizeBack(name='restore')([mask_out, img_raw] )\r\n",
    "\r\n",
    "    model = Model( inputs = img_raw, outputs = mask_out, name = 'QDL-CMFD')\r\n",
    "    if weight_file is not None :\r\n",
    "        try :\r\n",
    "            model.load_weights(weight_file)\r\n",
    "            print(\"Model successfully loaded from {}\".format(weight_file))\r\n",
    "        except Exception as e :\r\n",
    "            print(\"Failed to load from {} / Error: {}\".format(weight_file,e))\r\n",
    "    return model"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "def QDLCMFD_decoder( QDLCMFDmodel, rgb ) :\r\n",
    "    single_sample_batch = np.expand_dims( rgb, axis=0 )\r\n",
    "    pred = QDLCMFDmodel.predict( single_sample_batch )[0]\r\n",
    "    return pred\r\n",
    "\r\n",
    "def visualize_result( rgb, gt, pred, figsize=(12,4), title=None ) :\r\n",
    "    plt.figure( figsize=figsize )\r\n",
    "    plt.subplot(131)\r\n",
    "    plt.imshow( rgb )\r\n",
    "    plt.title('input image')\r\n",
    "    plt.subplot(132)\r\n",
    "    plt.title('ground truth')\r\n",
    "    plt.imshow(gt)\r\n",
    "    plt.subplot(133)\r\n",
    "    plt.imshow(pred)\r\n",
    "    plt.title('DCMFD pred')\r\n",
    "    if title is not None :\r\n",
    "        plt.suptitle( title )\r\n",
    "\r\n",
    "def visualize_1result( rgb, pred, figsize=(12,4), title=None ) :\r\n",
    "    plt.figure( figsize=figsize )\r\n",
    "    plt.subplot(121)\r\n",
    "    plt.imshow( rgb )\r\n",
    "    plt.title('input image')\r\n",
    "    plt.subplot(122)\r\n",
    "    plt.imshow(pred)\r\n",
    "    plt.title('DCMFD pred')\r\n",
    "    if title is not None :\r\n",
    "        plt.suptitle( title )"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "QDLCMFDmodel = create_QDLCMFD_test_model('./Model/QDL-CMFD-Model.hd5')"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "image = Image.open('./Images/SR.jpg')\r\n",
    "pred = QDLCMFD_decoder(QDLCMFDmodel, image)\r\n",
    "visualize_1result(image, pred)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "----\r\n",
    "\r\n",
    "**QDL-CMFD** v.1.2.16 | a Quality-independent and Deep Learning-based Copy-Move image Forgery Detection method.\r\n",
    "<br>{Image Forgery Detection, Copy-Move Forgery, Deep Learning, Generative Adversarial Networks, Convolutional Neural Networks, Super-Resolution Enhancement}\r\n",
    "\r\n",
    "© Proposed method implementation by **Mehrad Aria** for Paper [QDL-CMFD](https://doi.org/10.1016/j.neucom.2022.09.017).\r\n",
    "<br>September 2021 / Shiraz, Iran.\r\n",
    "\r\n",
    "----"
   ],
   "metadata": {}
  }
 ],
 "metadata": {
  "kernelspec": {
   "name": "python3",
   "display_name": "Python 3.8.11 64-bit (virtualenv)"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.11"
  },
  "interpreter": {
   "hash": "4881f5eacd7088bcf5a836f9e344288743d588393de6dc96f82d4b26fa31d287"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}