CNN_binary_mult_marion.py

# -*- coding: utf-8 -*-
"""embeddings_CNN.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/1vuhFeGaLZj88_oDR5v7pqMjFIgzRD3x6

## Import
"""

# this is for google colab
# from google.colab import drive
# drive.mount('/content/drive')

import numpy as np
import pandas as pd

from sklearn.metrics import classification_report
from gensim.models import Word2Vec, KeyedVectors
from sklearn.model_selection import train_test_split
import matplotlib.pyplot as plt

from keras.layers import *
from keras.preprocessing.sequence import pad_sequences
from keras.preprocessing.text import Tokenizer
from keras.utils import to_categorical
from keras.models import Model, Input
from keras.optimizers import Adam
from collections import defaultdict
import json
import random

random.seed(1612)

def read_corpus(corpus_file):
    """read input document and return the textual articles
    and either the bias or hyperpartisan label"""

    with open(corpus_file) as json_file:
        data = json.load(json_file)

    data = pd.DataFrame(data)
    documents = data.text
    labels_bin = data.hyperp
    labels_mult = data.bias
    titles = data.title

    #join titles and documents
    docs = []
    for t, d in zip(titles, documents):
        docs.append(t+d)

    docs = np.asarray(docs)

    return docs, labels_bin, labels_mult


def document_transform(Xtrain, Xtest):

    toki = Tokenizer(oov_token='UNK')
    toki.fit_on_texts(Xtrain)

    Xtrain_seq = toki.texts_to_sequences(Xtrain)
    Xtest_seq = toki.texts_to_sequences(Xtest)

    word2index = toki.word_index
    word2index['PAD'] = 0
    index2word = toki.index_word
    index2word[0] = 'PAD'

    # get max length of words (max 2500, mean 600)
    max_len = 1000

    Xtrain_pad = pad_sequences(Xtrain_seq, maxlen=max_len)
    Xtest_pad = pad_sequences(Xtest_seq, maxlen=max_len)

    return Xtrain_pad, Xtest_pad, word2index, index2word


def label_transform(Ytrain, Ytest):
    if len(set(Ytrain)) == 2:
        Ytrain = np.asarray([0. if l == False else 1. for l in Ytrain_bin])
        Ytest = np.asarray([0. if l == False else 1. for l in Ytest_bin])
    elif len(set(Ytrain)) > 2:
        mult_labels = set(Ytrain)

        label_dict = defaultdict()
        for i, l in enumerate(mult_labels):
            label_dict[l] = i

        # transform into list of numbers
        Ytrain = [label_dict[label] for label in Ytrain]
        Ytest = [label_dict[label] for label in Ytest]

        no_cls = len(label_dict)

        Ytrain = np.asarray([to_categorical(label, num_classes=no_cls) for label in Ytrain])
        Ytest = np.asarray([to_categorical(label, num_classes=no_cls) for label in Ytest])
    else:
        print("something went wrong")

    return Ytrain, Ytest


def load_embeddings(model, index2word, embed_len):
    index2embed = dict()

    for i, w in index2word.items():
        try:
            embed = model[w]
        except KeyError:
            #embed = model['UNK']
            embed = np.zeros(embed_len)
        index2embed[i] = embed

    return index2embed


def get_embedding_layer(word2index, index2embed, max_len):
    # compute embedding matrix
    embedding_matrix = np.zeros((len(word2index) + 1, embed_len))
    for word, i in word2index.items():
        embedding_vector = index2embed[i]
        if embedding_vector is not None:
            # words not found in embedding index will be all-zeros.
            embedding_matrix[i] = embedding_vector

    # load embedding matrix into embedding layer
    embedding_layer = Embedding(len(word2index) + 1,
                                embed_len,
                                weights=[embedding_matrix],
                                input_length=max_len,
                                trainable=False)
    return embedding_layer


def cnn(binary, embedding, lr=0.001, hidden=200):
    if binary:
        loss = 'binary_crossentropy'
        out = 1
        act_out = 'sigmoid'
    else:
        loss = 'categorical_crossentropy'
        out = 5
        act_out = 'softmax'

    filters = 100
    kernel_size = 3
    optim = Adam(lr=lr)

    sequence_input = Input(shape=(max_len,), dtype='int32')
    embedded_sequences = embedding(sequence_input)

    drop1 = Dropout(0.2)(embedded_sequences)

    conv1 = Conv1D(filters,
                   kernel_size,
                   padding='valid',
                   activation='relu',
                   strides=1)(drop1)

    pool = GlobalMaxPool1D()(conv1)

    dense1 = Dense(hidden, activation='relu')(pool)
    drop2 = Dropout(0.5)(dense1)
    output = Dense(out, activation=act_out)(drop2)
    model = Model(inputs=sequence_input, outputs=output)
    model.compile(loss=loss, optimizer=optim, metrics=['accuracy'])

    model.summary()

    return model


def plot_history(history):
    # Plot training & validation accuracy values
    plt.plot(history.history['accuracy'])
    plt.plot(history.history['val_accuracy'])
    plt.title('Model accuracy')
    plt.ylabel('Accuracy')
    plt.xlabel('Epoch')
    plt.legend(['Acc', 'Val_acc'], loc='upper left')
    plt.show()

    # Plot training & validation loss values
    plt.plot(history.history['loss'])
    plt.plot(history.history['val_loss'])
    plt.title('Model loss')
    plt.ylabel('Loss')
    plt.xlabel('Epoch')
    plt.legend(['Train', 'Val'], loc='upper left')
    plt.show()


def report(true, pred):
    # binary classification returns a matrix with shape (X,1),
    # multiclass returns shape (X,5)
    if pred.shape[1] == 1:
        pred_converted = [0 if p < 0.5 else 1 for p in pred]
    else:
        pred_converted = np.argmax(pred, axis=1)
        true = np.argmax(true, axis=1)
    print(classification_report(true, pred_converted))

    return

def train_test(data, indices):
    train_idx, test_idx = train_test_split(indices, test_size=0.2)
    train = data[train_idx]
    test = data[test_idx]

    return train, test

if __name__ == '__main__':
    # Import and split up documents and labels
    print('Import data..')
    X, Y_bin, Y_mult = read_corpus('data/tokenized_with_NUM.json')
    #X_big, Y_bin, Y_multX = read_corpus('data/tokenized_with_NUM_whole.json')

    indices = np.arange(len(X))
    train_idx, test_idx = train_test_split(indices, test_size=0.2)

    Xtrain = X[train_idx]
    Xtest = X[test_idx]

    Ytrain_bin = Y_bin[train_idx]
    Ytest_bin = Y_bin[test_idx]

    Ytrain_mult = Y_mult[train_idx]
    Ytest_mult = Y_mult[test_idx]

    print('Label and document transformation..')
    # Transform documents: text to sequence and padding
    Xtrain, Xtest, word2index, index2word = document_transform(Xtrain, Xtest)

    # Transform labels to binary float or categorical
    Ytrain_bin, Ytest_bin = label_transform(Ytrain_bin, Ytest_bin)
    Ytrain_mult, Ytest_mult = label_transform(Ytrain_mult, Ytest_mult)

    print('Shape of data tensor:', Xtrain.shape)
    print('Shape of binary label tensor:', Ytrain_bin.shape)
    print('Shape of multilabel tensor:', Ytrain_mult.shape)

    # Embedding

    print('Load and prepare embeddings..')

    embeddings = Word2Vec.load('data/model_t_t.bin')
    #embeddings = KeyedVectors.load_word2vec_format('data/GoogleNews-vectors-negative300.bin', binary=True)
    embed_len = 100
    max_len = Xtrain.shape[1]

    index2embed = load_embeddings(embeddings, index2word, embed_len)
    embedding_layer = get_embedding_layer(word2index, index2embed, max_len)

    #CNN Classifiers

    print('Fitting the binary model.. ')

    epochs = 10
    batch = 1024

    model_bin = cnn(binary=True, embedding=embedding_layer)
    history_bin = model_bin.fit(Xtrain, Ytrain_bin, batch_size=batch, epochs=epochs, verbose=1,
                                validation_split=0.1)
    plot_history(history_bin)
    pred_bin = model_bin.predict(Xtest)
    report(Ytest_bin, pred_bin)

    print('Fitting the multi-class model..')

    epochs = 10
    batch = 256

    model_mult = cnn(binary=False, embedding=embedding_layer)
    history_mult = model_mult.fit(Xtrain, Ytrain_mult, batch_size=batch, epochs=epochs, verbose=1,
                                  validation_split=0.1)
    plot_history(history_mult)
    pred_mult = model_mult.predict(Xtest)
    print(pred_mult.shape)

    report(Ytest_mult, pred_mult)