# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     name: python3
# ---

# + [markdown] id="view-in-github" colab_type="text"
# <a href="https://colab.research.google.com/github/maxwelljohn/siamese-word2vec/blob/master/siamese.ipynb" target="_parent"><img src="https://colab.research.google.com/assets/colab-badge.svg" alt="Open In Colab"/></a>

# + id="fv-b7O6mz7G6" colab_type="code" colab={}
import numpy as np

import datetime
import heapq
import itertools
import matplotlib.pyplot as plt
# %matplotlib inline
import nltk
import os
import random
import skimage.transform
import sys
import time
import unittest
nltk.download('wordnet')
nltk.download('punkt')
from nltk.corpus import wordnet as wn
from keras.utils.data_utils import get_file
from keras.models import Model
from keras.layers import Input, Flatten, Dense, Dropout, Lambda
from keras.optimizers import Adam
from keras import backend as K
from keras import regularizers
from keras.constraints import non_neg
from scipy.misc import imread
from sklearn.manifold import TSNE

# !pip install scikit-optimize==0.5.2
import skopt

# !pip install -U -q PyDrive==1.3.1
from pydrive.auth import GoogleAuth
from pydrive.drive import GoogleDrive
from google.colab import auth
from oauth2client.client import GoogleCredentials


# + id="85s8R9-WEyOb" colab_type="code" colab={}
def similarity(a, b):
    # Cosine similarity
    return np.dot(a, b) / max(np.linalg.norm(a) * np.linalg.norm(b), sys.float_info.epsilon)


def keras_norm(vect):
    return K.sqrt(K.batch_dot(vect, vect, axes=1))


def keras_similarity(vects):
    x, y = vects
    # Cosine similarity
    return K.batch_dot(x, y, axes=1) / K.maximum(keras_norm(x) * keras_norm(y), K.epsilon())


def sim_output_shape(shapes):
    shape1, shape2 = shapes
    return (shape1[0], 1)


def create_base_network(input_shape, output_size=128, reg_rate=0):
    '''Base network to be shared (eq. to feature extraction).
    '''
    input = Input(shape=input_shape)
    x = input
    x = Dense(output_size, kernel_regularizer=regularizers.l2(reg_rate))(x)
    return Model(input, x)


def syn_accuracy(y_true, y_pred):
    '''Compute synonym classification accuracy with a variable threshold on similarities.
    '''
    median = np.median(y_pred)
    pred = y_pred.ravel() > median
    return np.mean(pred == y_true)


def syn_accuracy_scores(y_true, y_pred):
    y_pred = y_pred.ravel()
    median = np.median(y_pred)
    sd = np.std(y_pred)
    standardized = (y_pred - median) / sd
    multiplier = y_true * 2 - 1
    return standardized * multiplier


def ant_accuracy(y_true, y_pred):
    '''Compute antonym classification accuracy with a variable threshold on similarities.
    '''
    median = np.median(y_pred)
    pred = -((y_pred.ravel() < median).astype(np.int, casting='safe', copy=False))
    return np.mean(pred == y_true)


def keras_syn_accuracy(y_true, y_pred):
    '''Compute synonym classification accuracy with a fixed threshold on similarities.
    '''
    return K.mean(K.equal(y_true, K.cast(y_pred > 0.75, y_true.dtype)))


def keras_ant_accuracy(y_true, y_pred):
    '''Compute antonym classification accuracy with a fixed threshold on similarities.
    '''
    return K.mean(K.equal(y_true, K.cast(y_pred < -0.75, y_true.dtype)))


# + id="CxHK8Dd1z4k1" colab_type="code" colab={}
# !unzip -u glove.6B.zip || (wget http://nlp.stanford.edu/data/glove.6B.zip && unzip -u glove.6B.zip)

# + id="lFVWMt_f0fWC" colab_type="code" colab={}
# !egrep '^[a-z]+ ' glove.6B.300d.txt > glove.tokens.txt
# !cut -d' ' -f1 glove.tokens.txt > glove.tokens.strings.txt
# !cut -d' ' -f2- glove.tokens.txt > glove.tokens.vectors.txt

# + id="Q2Q9l3PN110t" colab_type="code" colab={}
word_strings = np.loadtxt('glove.tokens.strings.txt', dtype=object)
word_vectors = np.loadtxt('glove.tokens.vectors.txt')
input_shape = word_vectors.shape[1:]
assert len(word_strings) == len(word_vectors)

index_for_word = {}
for i, word in enumerate(word_strings):
    index_for_word[word] = i

# + id="foWPgSNi-ihu" colab_type="code" colab={}
# !rm *.zip *.txt

# + id="-T84gxLzJU9B" colab_type="code" outputId="30cecf0f-50c5-49fd-cef1-e2c4e53b8569" colab={"base_uri": "https://localhost:8080/", "height": 102}
wnl = nltk.WordNetLemmatizer()


def might_be_synonyms(w1, w2):
    s1 = set()
    d1 = set()
    s2 = set()
    d2 = set()
    for synset in wn.synsets(w1):
        lemma_names = synset.lemma_names()
        s1.update(lemma_names)
        d1.update(nltk.word_tokenize(synset.definition()))
    for synset in wn.synsets(w2):
        lemma_names = synset.lemma_names()
        s2.update(lemma_names)
        d2.update(nltk.word_tokenize(synset.definition()))
    total_intersection = len(s1.intersection(s2)) + len(d1.intersection(s2)) + len(d2.intersection(s1))
    return total_intersection > 0


class TestSynonyms(unittest.TestCase):
    def test_true(self):
        pairs = [
            ('car', 'auto'),
            ('auto', 'car'),
            ('car', 'railcar'),
            ('small', 'tiny'),
            ('small', 'miniature'),
        ]
        for word1, word2 in pairs:
            self.assertTrue(might_be_synonyms(word1, word2))
    def test_false(self):
        pairs = [
            ('car', 'airplane'),
            ('car', 'fast'),
            ('small', 'focused'),
            ('small', 'accidental'),
            ('small', 'rabbit'),
            ('small', 'flippant'),
        ]
        for word1, word2 in pairs:
            self.assertFalse(might_be_synonyms(word1, word2))


def get_antonyms(word):
    result = set()
    for lemma in wn.lemmas(word):
        for antonym in lemma.antonyms():
            result.add(antonym.name())
    return result


def might_be_antonyms(w1, w2):
    w1_antonyms = get_antonyms(w1)
    w2_antonyms = get_antonyms(w2)
    for w1_antonym in w1_antonyms:
        if might_be_synonyms(w2, w1_antonym):
            return True
    for w2_antonym in w2_antonyms:
        if might_be_synonyms(w1, w2_antonym):
            return True


class TestAntonyms(unittest.TestCase):
    def test_get_antonyms(self):
        self.assertEqual(get_antonyms('big'), set(['little', 'small']))
        self.assertEqual(get_antonyms('fast'), set(['slow']))
        self.assertEqual(get_antonyms('big'), set(['little', 'small']))
    def test_true(self):
        pairs = [
            ('big', 'small'),
            ('big', 'minor'),
            ('big', 'tiny'),
            ('big', 'miniature'),
            ('fast', 'slow'),
            ('fast', 'sluggish'),
            ('loud', 'soft'),
            ('loud', 'quiet'),
        ]
        for word1, word2 in pairs:
            self.assertTrue(might_be_antonyms(word1, word2))
    def test_false(self):
        pairs = [
            ('big', 'huge'),
            ('big', 'clean'),
            ('fast', 'speedy'),
            ('fast', 'unusual'),
            ('loud', 'noisy'),
            ('loud', 'flat'),
        ]
        for word1, word2 in pairs:
            self.assertFalse(might_be_antonyms(word1, word2))


unittest.main(argv=['first-arg-is-ignored'], exit=False)


def create_pairs(word_strings, word_vectors, class_indices, pos_label, max_per_class=float('infinity')):
    '''Positive and negative pair creation.
    Alternates between positive and negative pairs.
    '''
    string_pairs = []
    vector_pairs = []
    labels = []
    for family in class_indices:
        sibling_pairs = np.array(list(itertools.combinations(family, 2)))

        shuffled_indices = np.arange(len(word_strings))
        np.random.shuffle(shuffled_indices)
        next_index = 0

        if len(sibling_pairs) > max_per_class:
            sibling_pairs = sibling_pairs[np.random.choice(len(sibling_pairs), max_per_class)]

        for sibling_pair in sibling_pairs:
            np.random.shuffle(sibling_pair)
            anchor, pos = sibling_pair
            string_pairs.append([word_strings[anchor], word_strings[pos]])
            vector_pairs.append([word_vectors[anchor], word_vectors[pos]])
            labels.append(pos_label)
            
            random_neg = None
            while random_neg == None or random_neg in family or \
                    might_be_synonyms(word_strings[anchor], word_strings[random_neg]) or \
                    might_be_antonyms(word_strings[anchor], word_strings[random_neg]):
                if next_index >= len(shuffled_indices):
                    np.random.shuffle(shuffled_indices)
                    next_index = 0
                random_neg = shuffled_indices[next_index]
                next_index += 1

            string_pairs.append([word_strings[anchor], word_strings[random_neg]])
            vector_pairs.append([word_vectors[anchor], word_vectors[random_neg]])
            labels.append(0)
    assert len(string_pairs) == len(vector_pairs) == len(labels)
    return np.array(string_pairs), np.array(vector_pairs), np.array(labels)


def create_train_dev_test(word_strings, word_vectors, class_count, word_class, pos_label):
    classes = list(range(class_count))
    random.seed(850101)
    random.shuffle(classes)
    random.seed()
    train_dev_split = round(len(classes)*0.6)
    dev_test_split = round(len(classes)*0.8)
    train_classes = classes[:train_dev_split]
    dev_classes = classes[train_dev_split:dev_test_split]
    test_classes = classes[dev_test_split:]
    assert len(set(train_classes).intersection(set(dev_classes))) == 0
    assert len(set(dev_classes).intersection(set(test_classes))) == 0
    assert len(set(train_classes).intersection(set(test_classes))) == 0

    class_indices = [np.where(word_class == c)[0] for c in train_classes]
    tr_strings, tr_pairs, tr_y = create_pairs(word_strings, word_vectors, class_indices, pos_label)

    class_indices = [np.where(word_class == c)[0] for c in dev_classes]
    dev_strings, dev_pairs, dev_y = create_pairs(word_strings, word_vectors, class_indices, pos_label)

    class_indices = [np.where(word_class == c)[0] for c in test_classes]
    te_strings, te_pairs, te_y = create_pairs(word_strings, word_vectors, class_indices, pos_label)
    
    return tr_strings, tr_pairs, tr_y, dev_strings, dev_pairs, dev_y, te_strings, te_pairs, te_y


# + id="x6A3h9Wj2Xs9" colab_type="code" colab={}
syn_class = -np.ones(len(word_strings), dtype=np.int)
shuffled_indices = np.arange(len(word_strings))
np.random.shuffle(shuffled_indices)
class_count = 0
for i in shuffled_indices:
    word = word_strings[i]
    synsets = wn.synsets(word)
    if synsets:
        syn_indices = [index_for_word[syn] for syn in synsets[0].lemma_names() if syn in index_for_word and syn_class[index_for_word[syn]] == -1]
        if len(syn_indices) > 1:
            syn_class[syn_indices] = class_count
            class_count += 1

tr_syn_strings, tr_syn_pairs, tr_syn_y, \
dev_syn_strings, dev_syn_pairs, dev_syn_y, \
te_syn_strings, te_syn_pairs, te_syn_y = create_train_dev_test(word_strings, word_vectors, class_count, syn_class, 1)

# + id="o13AAwvvaIlM" colab_type="code" colab={}
ant_class = -np.ones(len(word_strings), dtype=np.int)
shuffled_indices = np.arange(len(word_strings))
np.random.shuffle(shuffled_indices)
class_count = 0
for i in shuffled_indices:
    word = word_strings[i]
    antonyms = get_antonyms(word)
    if antonyms:
        known_antonyms = [a for a in antonyms if a in index_for_word]
        if known_antonyms:
            ant_indices = [i, index_for_word[known_antonyms[0]]]
            if all(ant_class[ant_indices] == -1):
                ant_class[ant_indices] = class_count
                class_count += 1

tr_ant_strings, tr_ant_pairs, tr_ant_y, \
dev_ant_strings, dev_ant_pairs, dev_ant_y, \
te_ant_strings, te_ant_pairs, te_ant_y = create_train_dev_test(word_strings, word_vectors, class_count, ant_class, -1)

# + id="q17UoTpZyNHX" colab_type="code" colab={"base_uri": "https://localhost:8080/", "height": 34} outputId="00d0d2df-f12f-4c6f-b98b-08358a6bf930"
len(tr_ant_pairs)


# + id="cbbPZrpLSVpd" colab_type="code" colab={}
def fit_model(tr_data, output_size, optimizer, batch_size, epochs, reg_rate, deviation_dropoff):
    # network definition
    base_network = create_base_network(input_shape, output_size, reg_rate)

    input_a = Input(shape=input_shape)
    input_b = Input(shape=input_shape)

    # because we re-use the same instance `base_network`,
    # the weights of the network
    # will be shared across the two branches
    processed_a = base_network(input_a)
    processed_b = base_network(input_b)

    distance = Lambda(keras_similarity,
                      output_shape=sim_output_shape)([processed_a, processed_b])

    model = Model([input_a, input_b], distance)

    def contrastive_sim_loss(y_true, y_pred):
        return K.mean(-y_true * y_pred +
                      K.cast(K.equal(y_true, 0), 'float32') * (K.abs(y_pred) ** deviation_dropoff))

    # train
    if optimizer == 'adam':
        opt = Adam()
    else:
        raise ValueError("unknown optimizer")
    model.compile(loss=contrastive_sim_loss, optimizer=opt)
    
    for tr_pairs, tr_y in tr_data:
        model.fit([tr_pairs[:, 0], tr_pairs[:, 1]], tr_y,
            batch_size=batch_size,
            epochs=epochs,
            verbose=0)

    return model

def eval_model(model):
    y_pred = model.predict([tr_syn_pairs[:, 0], tr_syn_pairs[:, 1]])
    syn_tr_acc = syn_accuracy(tr_syn_y, y_pred)
    y_pred = model.predict([dev_syn_pairs[:, 0], dev_syn_pairs[:, 1]])
    syn_dev_acc = syn_accuracy(dev_syn_y, y_pred)

    y_pred = model.predict([tr_ant_pairs[:, 0], tr_ant_pairs[:, 1]])
    ant_tr_acc = ant_accuracy(tr_ant_y, y_pred)
    y_pred = model.predict([dev_ant_pairs[:, 0], dev_ant_pairs[:, 1]])
    ant_dev_acc = ant_accuracy(dev_ant_y, y_pred)

    print("Accuracy:")
    print("|         | Training Set | Dev Set |")
    print("|--------:|-------------:|--------:|")
    print("| Synonym |      {:7.2%} | {:7.2%} |".format(syn_tr_acc, syn_dev_acc))
    print("| Antonym |      {:7.2%} | {:7.2%} |".format(ant_tr_acc, ant_dev_acc))
    return syn_tr_acc, syn_dev_acc, ant_tr_acc, ant_dev_acc

def map_word_vectors(word_vectors, model):
    base_network = model.layers[2]
    l1_weights = base_network.layers[1].get_weights()[0]
    l1_biases = base_network.layers[1].get_weights()[1]
    mapped = word_vectors @ l1_weights + l1_biases
    mapped_2 = (l1_weights.T @ word_vectors.T + l1_biases.reshape((-1, 1))).T
    assert np.isclose(mapped, mapped_2).all()
    return mapped


# + id="kPptTE4gShhV" colab_type="code" outputId="60c89469-d178-4004-92ab-9ca44cd4c869" colab={"base_uri": "https://localhost:8080/", "height": 119}
syn_tr_acc = syn_accuracy(
    tr_syn_y, np.array([similarity(p[0], p[1]) for p in tr_syn_pairs])
)
syn_dev_acc = syn_accuracy(
    dev_syn_y, np.array([similarity(p[0], p[1]) for p in dev_syn_pairs])
)
ant_tr_acc = ant_accuracy(
    tr_ant_y, np.array([similarity(p[0], p[1]) for p in tr_ant_pairs])
)
ant_dev_acc = ant_accuracy(
    dev_ant_y, np.array([similarity(p[0], p[1]) for p in dev_ant_pairs])
)

print('Baseline...')
print("Accuracy:")
print("|         | Training Set | Dev Set |")
print("|--------:|-------------:|--------:|")
print("| Synonym |      {:7.2%} | {:7.2%} |".format(syn_tr_acc, syn_dev_acc))
print("| Antonym |      {:7.2%} | {:7.2%} |".format(ant_tr_acc, ant_dev_acc))

# + id="l562zedZzM_N" colab_type="code" outputId="8a249ea5-d29a-47cb-b1c4-79db5373128f" colab={"base_uri": "https://localhost:8080/", "height": 442}
print('Training on synonyms...')
syn_model = fit_model([(tr_syn_pairs, tr_syn_y)], 140, 'adam', 256, 30, 1e-09, 6)
eval_model(syn_model)
print('Training on antonyms...')
ant_model = fit_model([(tr_ant_pairs, tr_ant_y)], 140, 'adam', 256, 30, 1e-09, 6)
eval_model(ant_model)
print('Training on synonyms & antonyms...')
both_model = fit_model([(np.vstack((tr_syn_pairs, tr_ant_pairs)), np.hstack((tr_syn_y, tr_ant_y)))], 140, 'adam', 256, 30, 1e-09, 6)
eval_model(both_model)
print('Training on synonyms, then antonyms...')
seq_model = fit_model([(tr_syn_pairs, tr_syn_y), (tr_ant_pairs, tr_ant_y)], 140, 'adam', 256, 30, 1e-09, 6)
eval_model(seq_model)
print('')

# + id="p1i_j4YWy2La" colab_type="code" colab={}
space = [
    skopt.space.Integer(1, 300, name='output_size'),
    skopt.space.Categorical(['adam'], name='optimizer'),
    skopt.space.Categorical([128, 256, 512], name='batch_size'),
    skopt.space.Categorical([30], name='epochs'),
    skopt.space.Real(0.000000001, 1000, prior='log-uniform', name='reg_rate'),
    skopt.space.Integer(1, 50, name='deviation_dropoff'),
]

@skopt.utils.use_named_args(space)
def objective(output_size, optimizer, batch_size, epochs, reg_rate, deviation_dropoff):
    start_time = datetime.datetime.now()
    model = fit_model([(tr_syn_pairs, tr_syn_y)], output_size, optimizer, batch_size, epochs, reg_rate, deviation_dropoff)

    print('* Optimization took {:.0f} seconds'.format((datetime.datetime.now() - start_time).total_seconds()))
    syn_tr_acc, syn_dev_acc, ant_tr_acc, ant_dev_acc = eval_model(model)
    print('')
    return -syn_dev_acc

# Authenticate and create the PyDrive client.
auth.authenticate_user()
gauth = GoogleAuth()
gauth.credentials = GoogleCredentials.get_application_default()
drive = GoogleDrive(gauth)

checkpoint_filename = "{}-checkpoint.pkl".format(time.asctime())
checkpoint_filepath = os.path.join(os.curdir, checkpoint_filename)

def backup(res):
    skopt.dump(res, checkpoint_filepath)
    uploaded = drive.CreateFile({'title': checkpoint_filename})
    uploaded.SetContentFile(checkpoint_filepath)
    uploaded.Upload()
    print('Uploaded file with ID {}'.format(uploaded.get('id')))

x0 = [270, 'adam', 256, 30, 0.01, 14]
res = skopt.gp_minimize(objective, space, x0=x0, n_calls=25, callback=[backup])
res.x


# + id="yce-IhGCnte7" colab_type="code" colab={}
def get_pair_indices_to_display():
    #baseline_sim = np.array([similarity(p[0], p[1]) for p in tr_syn_pairs])
    #baseline_scores = syn_accuracy_scores(tr_syn_y, baseline_sim)
    optimized_sim = syn_model.predict([tr_syn_pairs[:, 0], tr_syn_pairs[:, 1]])
    optimized_scores = syn_accuracy_scores(tr_syn_y, optimized_sim)
    display_scores = (tr_syn_y == 1) * optimized_scores
    best_for_display = heapq.nlargest(50, zip(display_scores, range(len(display_scores))))
    return [index for improvement, index in best_for_display]

def plot_synonym_pairs(word_vectors, pairs):
    added = set()
    strings = []
    vectors = []
    for word1, word2 in pairs:
        if word1 not in added and word2 not in added:
            for word in (word1, word2):
                added.add(word)
                strings.append(word)
                vectors.append(word_vectors[index_for_word[word]])

    tsne = TSNE(init='pca')
    new_values = tsne.fit_transform(vectors)

    plt.figure(figsize=(16, 16))
    for i in range(0, len(new_values), 2):
        random.seed(i * 293778 + 581580)
        color = (random.random(), random.random(), random.random())
        string1, string2 = strings[i:i+2]
        point1, point2 = new_values[i:i+2]
        plt.scatter(point1[0], point1[1], c=[color])
        plt.annotate(string1,
                     xy=(point1[0], point1[1]),
                     xytext=(2, 2),
                     textcoords='offset points',
                     ha='left',
                     va='bottom')
        plt.scatter(point2[0], point2[1], c=[color])
        plt.annotate(string2,
                     xy=(point2[0], point2[1]),
                     xytext=(-2, -2),
                     textcoords='offset points',
                     ha='right',
                     va='top')
    random.seed()
    plt.show()


# + id="PtAWEeseqOcS" colab_type="code" colab={}
indices_to_display = get_pair_indices_to_display()
synonym_pairs = tr_syn_strings[indices_to_display]
syn_mapped_word_vectors = map_word_vectors(word_vectors, syn_model)

plot_synonym_pairs(word_vectors, synonym_pairs)
plot_synonym_pairs(syn_mapped_word_vectors, synonym_pairs)

# + id="517bxkz7fTHO" colab_type="code" colab={}


StarcoderdataJupyterScriptsDedupFiltered