W10-6 A Civil War Network.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Preliminaries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import networkx as nx\n",
    "import matplotlib.pyplot as plt\n",
    "import nltk"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def check_float(potential_float):\n",
    "    try:\n",
    "        float(potential_float)\n",
    "        return True\n",
    "    except ValueError:\n",
    "        return False\n",
    "\n",
    "def round_if_float(v, prec=3):\n",
    "    if check_float(v):\n",
    "        return round(float(v), prec)\n",
    "    return v\n",
    "\n",
    "from IPython.core.display import display, HTML\n",
    "def list_table(the_list, color_nums=False):\n",
    "    html = [\"<table style= 'border: 1px solid black; display:inline-block'>\"]\n",
    "    for row in the_list:\n",
    "        html.append(\"<tr>\")\n",
    "        for col in row:\n",
    "            if color_nums and check_float(col) and not float(col) == 0:\n",
    "                html.append(\"<td align='left' style='border: .5px solid gray; color: {1}; font-weight: bold'>{0}</td>\".format(round_if_float(col), color_nums))\n",
    "            else:\n",
    "                html.append(\"<td align='left' style='border: .5px solid gray;'>{0}</td>\".format(round_if_float(col)))\n",
    "        html.append(\"</tr>\")\n",
    "    html.append(\"</table>\")\n",
    "    return display(HTML(''.join(html)))\n",
    "\n",
    "def show_labeled_table(mat, col_names=None, row_names=None, nrows=10, ncols=10, color_nums=\"red\"):\n",
    "    sml = mat[:nrows, :ncols]\n",
    "    if col_names is not None:\n",
    "        sml = np.vstack([col_names[:ncols], sml])\n",
    "    if row_names is not None:\n",
    "        rnames = [[p] for p in row_names[:nrows]]\n",
    "        if col_names is not None:\n",
    "            new_col = np.array([[\"_\"]] + rnames)\n",
    "        else:\n",
    "            new_col = np.array(rnames)\n",
    "        sml = np.hstack((new_col, sml))\n",
    "    return list_table(sml, color_nums)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Network of relations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import wikipediaapi\n",
    "from string import punctuation\n",
    "from itertools import combinations\n",
    "import nltk\n",
    "import numpy as np\n",
    "from IPython.display import display, clear_output"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Read in the civil war corpus"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import wikipediaapi\n",
    "pages = [\n",
    "    \"American Civil War\",\n",
    "    \"Abraham Lincoln\",\n",
    "    \"Slavery in the United States\",\n",
    "    \"Slave states and free states\",\n",
    "    \"Emancipation Proclamation\",\n",
    "    \"Robert E. Lee\",\n",
    "    \"Ulysses S. Grant\",\n",
    "    \"Conclusion of the American Civil War\",\n",
    "    \"Origins of the American Civil War\",\n",
    "    \"Issues of the American Civil War\"\n",
    "]\n",
    "import re\n",
    "\n",
    "def underscorize(pagename):\n",
    "    return re.sub(\" \", \"_\", pagename)\n",
    "\n",
    "wiki_wiki = wikipediaapi.Wikipedia(\n",
    "        language='en',\n",
    "        extract_format=wikipediaapi.ExtractFormat.WIKI\n",
    ")\n",
    "page_dict = {}\n",
    "for page in pages:\n",
    "    pagename = underscorize(page)\n",
    "    print(pagename)\n",
    "    p_wiki = wiki_wiki.page(pagename)\n",
    "    page_text = p_wiki.text.split(\"\\n\")\n",
    "    page_paras = [para for para in page_text if len(para) > 1]\n",
    "    page_dict[pagename] = page_paras"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Extract the named entities"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Tag every word"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def tag_paragraph_sentences(para):\n",
    "    sentences = nltk.sent_tokenize(para)\n",
    "    tagged_sentences = []\n",
    "    for sent in sentences:\n",
    "        tokenized_sentence = nltk.word_tokenize(sent)\n",
    "        tagged_sentence = nltk.pos_tag(tokenized_sentence)\n",
    "        tagged_sentences.append(tagged_sentence)\n",
    "    return tagged_sentences\n",
    "tagged_sentences = []\n",
    "for name, page in page_dict.items():\n",
    "    for para in page:\n",
    "        tagged_sentences += tag_paragraph_sentences(para)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Identify the chunks using the nltk named-entity chunker"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "chunked_sentences = []\n",
    "for n, sent in enumerate(tagged_sentences):\n",
    "    if n % 500 == 0:\n",
    "        clear_output(wait=True)\n",
    "        print('Sentence {} of {}'.format(n, len(tagged_sentences)))\n",
    "    chunked_sentences.append(nltk.ne_chunk(sent))\n",
    "clear_output(wait=True)\n",
    "print(\"done\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def entity_to_tuple(t):\n",
    "    return tuple([t.label(), \" \".join([token for token, pos in t.leaves()])])\n",
    "\n",
    "def extract_entities(chunked_sentence):\n",
    "    entities = []\n",
    "    for i in chunked_sentence:\n",
    "        if type(i) == nltk.Tree:\n",
    "            entities.append(entity_to_tuple(i))\n",
    "    return entities\n",
    "\n",
    "def tuple_to_string(t):\n",
    "    return t[1] + \"_\" + t[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "extract_entities(chunked_sentences[0])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Find the most common entities.\n",
    "\n",
    "Also create a list that has the list of entities in each sentence."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "entity_fdist = nltk.FreqDist()\n",
    "ents_list = []\n",
    "for sent in chunked_sentences:\n",
    "    ents = extract_entities(sent)\n",
    "    ents_list.append(ents)\n",
    "    entity_fdist.update(ents)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "entity_fdist.most_common(25)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Build the adjacency matrix"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Our \"vocabulary\" will consist of the most common entities.\n",
    "\n",
    "Then we'll compute a vector for each sentence. The elements of the vector will be the count of how many times each of the entities in this vocabulary appeared.\n",
    "\n",
    "We'll use the vectors to build a termxdocument matrix. Then we'll convert this to a termxterm matrix, which we'll use as our adjacency matrix."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def compute_vector(words, vocab):\n",
    "    new_vector = []\n",
    "    for w in vocab:\n",
    "        tf = words.count(w)\n",
    "        new_vector.append(tf)\n",
    "    return np.array(new_vector)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "vocab_size = 100\n",
    "medium_vocab = [w[0] for w in entity_fdist.most_common(vocab_size)]\n",
    "doc_vectors = {}\n",
    "N = len(ents_list)\n",
    "entity_vector_list = []\n",
    "for ents in ents_list:\n",
    "    entity_vector_list.append(compute_vector(ents, medium_vocab))\n",
    "td_matrix = np.zeros([len(medium_vocab), len(entity_vector_list)])\n",
    "i = 0\n",
    "for entity_vector in entity_vector_list:\n",
    "    td_matrix[:, i] = entity_vector\n",
    "    i = i + 1\n",
    "for r in range(len(medium_vocab)):\n",
    "    td_matrix[r, :] = td_matrix[r, :]\n",
    "tt_matrix = np.dot(td_matrix, td_matrix.transpose())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "headings = [tup[1] for tup in medium_vocab]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "show_labeled_table(tt_matrix, headings, headings, nrows=25, ncols=25, color_nums=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def build_weighted_graph_from_Aij(am, vocab):\n",
    "    g = nx.Graph()\n",
    "    entity_strings = [tuple_to_string(tup) for tup in vocab]\n",
    "    for entity_data in vocab:\n",
    "        g.add_node(tuple_to_string(entity_data), type=entity_data[0], name=entity_data[1])\n",
    "    dim = len(am)\n",
    "    for r in range(dim):\n",
    "        for c in range(dim):\n",
    "            if r != c and am[r][c] != 0:\n",
    "                g.add_weighted_edges_from([(entity_strings[r], entity_strings[c], am[r][c])])\n",
    "    return g"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cw_g = build_weighted_graph_from_Aij(tt_matrix, vocab)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Draw the network"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's give different colors to each node depending on which type of entity it is."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "label_dict = {}\n",
    "for i, n in enumerate(cw_g.nodes):\n",
    "    label_dict[n] = vocab[i][1]\n",
    "weights = [cw_g[u][v]['weight'] / 5 for u, v in cw_g.edges()]\n",
    "color_dict = {\"PERSON\": \"blue\",\n",
    "             \"GPE\": \"green\",\n",
    "             \"ORGANIZATION\": \"red\",\n",
    "             \"LOCATION\": \"orange\",\n",
    "             \"GSP\": \"gray\"}\n",
    "node_colors = []\n",
    "for n in cw_g.nodes():\n",
    "    node_colors.append(color_dict[cw_g.nodes()[n][\"type\"]])\n",
    "    pos = nx.spring_layout(cw_g, k=2)\n",
    "plt.figure(figsize=(30, 30))\n",
    "plt.axis('off')\n",
    "nx.draw_networkx(cw_g, pos=pos, labels=label_dict, node_color=node_colors, width=weights, alpha = .5, font_size = 20, node_size = 500)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Export the network to a file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "nx.write_gexf(cw_g, \"civil_war.gexf\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.7"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": true
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}