GAN Model.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import common\n",
    "versePairs = common.loadTrainingData()\n",
    "X, Y = common.cleanAndSplitVerses(versePairs)\n",
    "Xenc, Yenc = common.encXandY(X, Y)\n",
    "# Use this to check that regexClean and regexUnclean are perfect inverses on the training data.\n",
    "# If the result is non empty you probably need to clean up the corresponding lines of your training set.\n",
    "assert len([v for v in versePairs if common.regexUnclean(common.regexClean(v)) != v]) == 0\n",
    "maxlen = 100\n",
    "Xnp, Ynp = common.padXandY(Xenc, Yenc, maxlen)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from keras import backend as K\n",
    "# Define the model. Uncomment this if you don't have the saved model available.\n",
    "from keras.models import Sequential, Model\n",
    "from keras.layers import LSTM, GRU, Dense, TimeDistributed, Bidirectional, Input, Embedding\n",
    "from keras.layers.merge import Concatenate\n",
    "from keras.layers.core import Dropout\n",
    "\n",
    "# Generator\n",
    "\n",
    "inputs = Input(shape=(100, 31))\n",
    "bidi = Bidirectional(LSTM(256, return_sequences=True))(inputs)\n",
    "den = Dense(4, activation='softmax')(bidi)\n",
    "gen = Model(inputs=inputs, outputs=den, name='gen_out')\n",
    "gen.compile(loss='categorical_crossentropy', optimizer='adam', sample_weight_mode='temporal', metrics=['categorical_accuracy'])\n",
    "#from keras.models import load_model\n",
    "#model = load_model(\"saved_1step_model\")\n",
    "\n",
    "gen.summary()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Discriminator\n",
    "discIn = Input((100, 35))\n",
    "disc = Bidirectional(GRU(128, return_sequences=True))(discIn)\n",
    "disc = GRU(64, return_sequences=True)(disc)\n",
    "disc = GRU(32)(disc)\n",
    "disc = Dense(1, activation='sigmoid')(disc)\n",
    "disc = Model(inputs=discIn, outputs=disc, name='disc_out')\n",
    "disc.compile(loss='binary_crossentropy', optimizer='adam', metrics=['accuracy'])\n",
    "disc.summary()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "common.make_trainable(disc, False)\n",
    "gan_input = Input((100,31))\n",
    "H = gen(gan_input)\n",
    "conc = Concatenate()([gan_input, H])\n",
    "gan_V = disc(conc)\n",
    "gan = Model(gan_input, outputs=[H, gan_V])\n",
    "gan.compile(loss='binary_crossentropy', optimizer='adam', metrics=['accuracy'])\n",
    "gan.summary()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Generate a weighting for the different characters. We want to penalise the model according to how rare a symbol is\n",
    "# so that rare symbols are more important to place correctly than common ones.\n",
    "import numpy as np\n",
    "from collections import Counter\n",
    "Yclass = np.argmax(Ynp, axis=2)\n",
    "chars = Counter(\"\".join(Y))\n",
    "freq = chars.values()\n",
    "total = np.sum(list(freq))\n",
    "toReplace = {}\n",
    "toReplace[0] = total/chars[\"0\"]\n",
    "toReplace[1] = total/chars[\"|\"]\n",
    "toReplace[2] = total/chars[\"·\"]\n",
    "toReplace[3] = total/chars[\"*\"]\n",
    "def replace(clas):\n",
    "    return toReplace[clas]\n",
    "sample_weight = np.vectorize(replace)(Yclass)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# A callback to display a particular verse after each epoch.\n",
    "from keras.callbacks import Callback\n",
    "class ShowVerse(Callback):\n",
    "    def __init__(self, verse=0):\n",
    "        self.verse = verse\n",
    "    def on_epoch_end(self, batch, logs={}):\n",
    "        pred = self.model.predict_classes(Xnp[self.verse:self.verse+1])\n",
    "        toComb = common.decClasses(pred[0])\n",
    "        print(common.regexUnclean(common.mergeStrings(X[self.verse], toComb)))\n",
    "        pred = self.model.predict(Xnp[self.verse:self.verse+1], batch_size=256)\n",
    "        toComb = common.getToComb(pred[0])\n",
    "        print(common.regexUnclean(common.mergeStrings(X[self.verse], toComb)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from keras.callbacks import Callback\n",
    "class EarlyStoppingByAccuracy(Callback):\n",
    "    def __init__(self, monitor='acc', value=0.6, verbose=0):\n",
    "        super(Callback, self).__init__()\n",
    "        self.monitor = monitor\n",
    "        self.value = value\n",
    "        self.verbose = verbose\n",
    "\n",
    "    def on_epoch_end(self, epoch, logs=None):\n",
    "        current = logs.get(self.monitor)\n",
    "        if current is None:\n",
    "            print(\"Early stopping requires %s available!\" % self.monitor, RuntimeWarning)\n",
    "\n",
    "        elif current > self.value:\n",
    "            if self.verbose > 0:\n",
    "                print(\"Epoch %05d: early stopping THR\" % epoch)\n",
    "            self.model.stop_training = True\n",
    "\n",
    "class EarlyStoppingByLoss(Callback):\n",
    "    def __init__(self, monitor='loss', value=0.5, verbose=0):\n",
    "        super(Callback, self).__init__()\n",
    "        self.monitor = monitor\n",
    "        self.value = value\n",
    "        self.verbose = verbose\n",
    "\n",
    "    def on_epoch_end(self, epoch, logs=None):\n",
    "        current = logs.get(self.monitor)\n",
    "        if current is None:\n",
    "            print(\"Early stopping requires %s available!\" % self.monitor, RuntimeWarning)\n",
    "\n",
    "        elif current < self.value:\n",
    "            if self.verbose > 0:\n",
    "                print(\"Epoch %05d: early stopping THR\" % epoch)\n",
    "            self.model.stop_training = True"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# A regex that matches a more-or-less correctly pointed verse\n",
    "regex = r\"[^|·*]+\\|[^|·*]+·*[^|·*]+\\|[^|·*]+\\*\\n[^|·*]+\\|[^|·*]+·*[^|·*]+\\|[^|·*]+·*[^|·*]+\\|[^|·*]+$\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "true = np.concatenate([Xnp, Ynp], axis=-1)\n",
    "def pre_train_gen(epochs):\n",
    "    common.make_trainable(gen, True)\n",
    "    gen.fit(Xnp, Ynp, epochs=epochs, batch_size=256, sample_weight=sample_weight)\n",
    "    common.make_trainable(gen, False)\n",
    "\n",
    "def train_disc(epochs, up_to_value=None):\n",
    "    gen_pred = gen.predict(Xnp)\n",
    "    false = np.concatenate([Xnp, gen_pred], axis=-1)\n",
    "    false_len = false.shape[0]\n",
    "    false_goal = np.zeros(false_len)\n",
    "    for i in range(false_len):\n",
    "        toComb = getClasses(gen_pred[i])\n",
    "        candidate = regexUnclean(mergeStrings(X[i], toComb))\n",
    "        if re.match(regex, candidate):\n",
    "            false_goal[i] = 1\n",
    "    true_then_false = np.concatenate([true, false], axis=0)\n",
    "    true_len = true.shape[0]\n",
    "    goal = np.concatenate([np.ones(true_len), false_goal])\n",
    "    common.make_trainable(disc, True)\n",
    "    if up_to_value:\n",
    "        disc.fit(true_then_false, goal, epochs=epochs, batch_size=256, callbacks=[EarlyStoppingByLoss(value=up_to_value)])\n",
    "    else:\n",
    "        disc.fit(true_then_false, goal, epochs=epochs, batch_size=256)\n",
    "    common.make_trainable(disc, False)\n",
    "\n",
    "from keras.callbacks import ModelCheckpoint\n",
    "    \n",
    "def train_gen(epochs, up_to_value=None, i=0):\n",
    "    common.make_trainable(gen, True)\n",
    "    if up_to_value:\n",
    "        gan.fit(Xnp, [Ynp, np.ones(true.shape[0])], epochs=epochs, batch_size=256, \n",
    "                callbacks=[ModelCheckpoint('weights.{0}.{1}'.format(i, '{epoch:02d}'), monitor='disc_out_loss'),\n",
    "                    EarlyStoppingByLoss(monitor=\"disc_out_loss\",value=up_to_value)])\n",
    "    else:\n",
    "        gan.fit(Xnp, [Ynp, np.ones(true.shape[0])], epochs=epochs, batch_size=256, \n",
    "                callbacks=[ModelCheckpoint('weights.{0}.{1}'.format(i, '{epoch:02d}'), monitor='disc_out_loss')])\n",
    "    common.make_trainable(gen, False)\n",
    "\n",
    "def train_both(epochs):\n",
    "    for i in range(epochs):\n",
    "        print(\"Epoch is {0}\".format(i))\n",
    "        print(\"Training discriminator\")\n",
    "        train_disc(100, up_to_value=0.2)\n",
    "        print(\"Training generator\")\n",
    "        train_gen(1000, up_to_value=0.2, i=i)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "pre_train_gen(500)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "gen.load_weights(\"gen_weights\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "gen.save_weights('gen_weights')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "train_both(30)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "train_disc(100)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "train_gen(100)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pred = gen.predict(Xnp, batch_size=256)\n",
    "tf = disc.predict(np.concatenate([Xnp, pred], axis=-1))\n",
    "for i in range(pred.shape[0]):\n",
    "    toComb = common.getClasses(pred[i])\n",
    "    print(tf[i])\n",
    "    print(common.regexUnclean(common.mergeStrings(X[i], toComb)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "pred = model.predict_classes(Xnp, batch_size=256)\n",
    "for i in range(pred.shape[0]):\n",
    "    toComb = decClasses(pred[i])\n",
    "    print(regexUnclean(mergeStrings(X[i], toComb)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "pred = model.predict(Xnp, batch_size=256)\n",
    "for i in range(pred.shape[0]):\n",
    "    toComb = getToComb(pred[i])\n",
    "    print(regexUnclean(mergeStrings(X[i], toComb)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Load and predict on some test data.\n",
    "from keras.preprocessing.sequence import pad_sequences\n",
    "test = []\n",
    "with open(\"testCleaned.txt\", 'r', encoding=\"utf-8\") as file:\n",
    "    for line in file:\n",
    "        test.append(line[:-1])\n",
    "testEnc = [common.encString(l) for l in test]\n",
    "testNp = pad_sequences(testEnc, maxlen=191)\n",
    "testPred = gen.predict_classes(testNp, batch_size=256)\n",
    "for i in range(len(test)):\n",
    "    toComb = common.decClasses(testPred[i])\n",
    "    print(common.regexUnclean(common.mergeStrings(test[i], toComb)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Load and predict on some test data.\n",
    "from keras.preprocessing.sequence import pad_sequences\n",
    "test = []\n",
    "with open(\"john1.txt\", 'r', encoding=\"utf-8\") as file:\n",
    "    for line in file:\n",
    "        test.append(line[:-1])\n",
    "testEnc = [encString(l) for l in test]\n",
    "testNp = pad_sequences(testEnc, maxlen=maxlen)\n",
    "testPred = gen.predict(testNp, batch_size=256)\n",
    "matches = 0\n",
    "for i in range(len(test)):\n",
    "    toComb = getClasses(testPred[i])\n",
    "    candidate = regexUnclean(mergeStrings(test[i], toComb))\n",
    "    if re.match(regex, candidate):\n",
    "        matches += 1\n",
    "        print(candidate)\n",
    "print(\"There were {0} verses and {1} matches.\".format(len(test), matches))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "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.5.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}