Downsampler.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import csv\n",
    "import os \n",
    "import sys\n",
    "import pdb\n",
    "import string"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Chris's Rhyming Scorer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# imports\n",
    "import re\n",
    "from pathlib import Path\n",
    "from typing import List, Union\n",
    "\n",
    "import numpy as np\n",
    "import pronouncing\n",
    "\n",
    "from youshen.util import calculate_edit_distance, clean\n",
    "\n",
    "LIMERICK_PATTERN = [[0, 1], [2, 3], [0, 4]]\n",
    "BLACKLIST =[\n",
    "    \"=+.*=+\",\n",
    "    \"(<.endoftext)*.>\"\n",
    "]\n",
    "\n",
    "\n",
    "def get_top_limerick(text: str):\n",
    "    \"\"\"\n",
    "    Parameters\n",
    "    ----------\n",
    "    text: Str\n",
    "        Samples of text generated by GPT-2\n",
    "\n",
    "    Returns\n",
    "    ----------\n",
    "    top_limerick: str\n",
    "        The best rhyming limerick from a set of samples generated by GPT-2\n",
    "\n",
    "    Examples\n",
    "    ----------\n",
    "    >>> from pathlib import Path\n",
    "    >>> from youshen.poem import get_top_limerick\n",
    "    >>> samples = Path(\"samples/samples.txt\")\n",
    "    >>> with open(samples) as file:\n",
    "    >>>     text = file.read()\n",
    "    >>> top = get_top_limerick(text)\n",
    "    \"\"\"\n",
    "    poem_samples = text.split(\"<|endoftext|>\")\n",
    "\n",
    "    poems = [\n",
    "        clean(poem_sample, BLACKLIST) \n",
    "        for poem_sample in poem_samples \n",
    "        if len(poem_sample) > 0\n",
    "    ]\n",
    "\n",
    "    poems = [poem for poem in poems if len(poem) > 0]\n",
    "    grouped_verses = [[line for line in poem.splitlines() if line] for poem in poems ]\n",
    "    poem_text = [\"\\n\".join(grouped_verse) for grouped_verse in grouped_verses if len(grouped_verse)>4] \n",
    "    poems = [SamplePoem(text=poem_sample, rhyme_patterns=LIMERICK_PATTERN, verse_length=5) \n",
    "                 for poem_sample in poem_text if len(poem_sample) >4]\n",
    "    indexed_scores = {poems.index(poem): poem.get_rhyme_score() for poem in poems}\n",
    "    top_poem_index = max(indexed_scores)\n",
    "    top_limerick = poems[top_poem_index].lines\n",
    "    return \"\\n\".join(top_limerick)\n",
    "\n",
    "class Limerick:\n",
    "    def __init__(self, lines: str, rhyme_patterns: List, max_length: int = None):\n",
    "        self.verse_lines = lines\n",
    "        if max_length:\n",
    "            self.verse_lines = self.verse_lines[0:max_length]\n",
    "#         pdb.set_trace()\n",
    "        try:\n",
    "            self.last_words = [line.split()[-1] for line in self.verse_lines] # Error here\n",
    "        except:\n",
    "            pdb.set_trace()\n",
    "        self.last_word_rhyming_part_pairs = {\n",
    "            word: self.__get_rhyming_parts(word) for word in self.last_words\n",
    "        }\n",
    "        self.rhyme_patterns = rhyme_patterns\n",
    "\n",
    "    def __get_phonemes(self, text: Union[str, List]):\n",
    "        \"\"\"returns all possible pronunciation of a word as phonemes\n",
    "        Language used: American English. Style: Arpabet\n",
    "        \"\"\"\n",
    "        if type(text) == str:\n",
    "            phonemes = pronouncing.phones_for_word(text)\n",
    "        else:\n",
    "            phonemes = [pronouncing.phones_for_word(word) for word in text]\n",
    "        return phonemes\n",
    "\n",
    "    def __get_rhyming_parts(self, word: str):\n",
    "        phonemes = self.__get_phonemes(word)\n",
    "        rhyming_parts = [pronouncing.rhyming_part(phoneme) for phoneme in phonemes]\n",
    "        return rhyming_parts\n",
    "\n",
    "    def __get_valid_rhyme_patterns(self):\n",
    "        valid_patterns = [\n",
    "            pattern\n",
    "            for pattern in self.rhyme_patterns\n",
    "            if not any(i > len(self.verse_lines) - 1 for i in pattern)\n",
    "        ]\n",
    "        return valid_patterns\n",
    "\n",
    "    def score(self, line_pair: List):\n",
    "        first_word = self.last_words[line_pair[0]]\n",
    "        second_word = self.last_words[line_pair[1]]\n",
    "        first_word_rhymes = self.__get_rhyming_parts(first_word)\n",
    "        second_word_rhymes = self.__get_rhyming_parts(second_word)\n",
    "        rhyme_score = 0\n",
    "        for first_word_rhyme in first_word_rhymes:\n",
    "            for second_word_rhyme in second_word_rhymes:\n",
    "                is_rhyming = first_word_rhyme == second_word_rhyme\n",
    "                if is_rhyming:\n",
    "                    rhyme_score = 1\n",
    "                    status = \"successfully matched\"\n",
    "                else:\n",
    "                    status = \"could not match\"\n",
    "                # uncomment to debug\n",
    "                print(\n",
    "                    f\" {status} -> {first_word}({first_word_rhyme}) and {second_word}({second_word_rhyme})\"\n",
    "                )\n",
    "        return int(rhyme_score)\n",
    "\n",
    "    def get_rhyme_score(self):\n",
    "        \"\"\"returns a rhyming score for the poem between 0 and 1.\n",
    "        \"\"\"\n",
    "        valid_patterns = self.__get_valid_rhyme_patterns()\n",
    "        scores = [self.score(pattern) for pattern in valid_patterns]\n",
    "        if len(scores) == 0:\n",
    "            return 0\n",
    "            # pdb.set_trace()\n",
    "        return sum(scores) / len(scores)\n",
    "\n",
    "    def __repr__(self):\n",
    "        return repr(\"\\n\".join(self.verse_lines))\n",
    "\n",
    "\n",
    "class SamplePoem:\n",
    "    def __init__(\n",
    "        self, text: str, rhyme_patterns: List, verse_length: int, blacklist: List = None\n",
    "    ):\n",
    "#         pdb.set_trace()\n",
    "        self.lines = [line for line in text.splitlines() if line]\n",
    "        self.verse_length = verse_length\n",
    "        self.rhyme_patterns = rhyme_patterns\n",
    "        intervals = list(range(0, len(self.lines), verse_length))\n",
    "        verse_lines_list = [self.lines[x : x + 5] for x in intervals]\n",
    "        self.verses = [\n",
    "            Limerick(\n",
    "                lines=verse_lines,\n",
    "                rhyme_patterns=self.rhyme_patterns,\n",
    "                max_length=self.verse_length,\n",
    "            )\n",
    "            for verse_lines in verse_lines_list\n",
    "        ]\n",
    "\n",
    "    def __get_item__(self, key):\n",
    "        return self.verses[key]\n",
    "\n",
    "    def get_rhyme_score(self):\n",
    "        if len(self.verses):\n",
    "            scores = [verse.get_rhyme_score() for verse in self.verses]\n",
    "            score = sum(scores) / len(scores)\n",
    "        else:\n",
    "            score = None\n",
    "        return score\n",
    "\n",
    "    def __repr__(self):\n",
    "        return repr(self.verses)\n",
    "\n",
    "\n",
    "def read_poems(file_path: Path, blacklist):\n",
    "    \"\"\"reads a file containing poems and returns a list of limerick samples found in the file\n",
    "    \"\"\"\n",
    "    with open(file_path) as file:\n",
    "        text = file.read()\n",
    "    poem_samples = text.split(\"<|endoftext|>\")\n",
    "    poems = [clean(sample, blacklist) for sample in poem_samples if len(sample) > 0]\n",
    "    return [poem for poem in poems if len(poem) > 0]\n",
    "\n",
    "\n",
    "def score_poems(file_path: Path, last_word_pattern: str, blacklist: List):\n",
    "    \"\"\"Reads limericks in generated samples and scores them between 0 and 1\n",
    "    \"\"\"\n",
    "    poems = [\n",
    "        SamplePoem(text=poem_sample, rhyme_patterns=LIMERICK_PATTERN, verse_length=5)\n",
    "        for poem_sample in read_poems(file_path, blacklist=blacklist)\n",
    "        if len(poem_sample) > 0\n",
    "    ]\n",
    "    poems = [poem for poem in poems if poem.lines]\n",
    "    poem_scores = [poem.get_rhyme_score() for poem in poems]\n",
    "    return poem_scores\n",
    "\n",
    "\n",
    "def test_scoring_limerick(sample_rhyme: Path, limerick_pattern: List):\n",
    "    \"\"\"Sanity check to test scoring of a single limerick\n",
    "    \"\"\"\n",
    "    with open(sample_rhyme) as rhyme_sample:\n",
    "        sample_corpus = rhyme_sample.read()\n",
    "    limerick_lines = [line for line in sample_corpus.splitlines() if line]\n",
    "    limerick = Limerick(\n",
    "        lines=limerick_lines, rhyme_patterns=limerick_pattern, max_length=5\n",
    "    )\n",
    "    print(\"Scoring limerick...\")\n",
    "    score = limerick.get_rhyme_score()\n",
    "    print(f\"Rhyme score is {score}\")\n",
    "    assert type(score) == float\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Sperate Text and Score"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "filename = \"samples_singh.txt\"\n",
    "limericks = []\n",
    "limerick = []\n",
    "with open(filename, \"r\") as file:\n",
    "    \n",
    "    for x in file:\n",
    "        \n",
    "        if \"======================================== \" in x:\n",
    "            continue \n",
    "        elif \"<|endoftext|>\" in x :\n",
    "            if len(limerick) == 5:\n",
    "                limericks.append(limerick)\n",
    "            limerick = []\n",
    "            continue\n",
    "        elif x == '\\n' or x == ' \\n':\n",
    "            continue\n",
    "        \n",
    "#         x = x.replace('\\n', '')\n",
    "        limerick.append(x)\n",
    "            \n",
    "poems = [SamplePoem(text=''.join(poem_sample), rhyme_patterns=LIMERICK_PATTERN, verse_length=5) \n",
    "         for poem_sample in limericks if len(poem_sample) >4]\n",
    "indexed_scores = {poems.index(poem): poem.get_rhyme_score() for poem in poems} "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "inds = [k for k,v in indexed_scores.items() if v == 1.0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "out = [limericks[i] for i in inds]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "filename = \"best_limericks.txt\"\n",
    "with open(filename, \"w\") as file:\n",
    "    for lim in out:\n",
    "        file.write(''.join(lim)+'\\r\\n')\n",
    "#         pdb.set_trace()"
   ]
  }
 ],
 "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.7.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}