lodes3_sandiego_subset.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Human Mobility in San Diego County\n",
    "Jessica Embury\n",
    "\n",
    "### Subset LODES data to only include San Diego block groups and aggregate to Block Groups\n",
    "### Excess tables deleted due to technical limitations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#import needed libraries\n",
    "from bs4 import BeautifulSoup\n",
    "import requests\n",
    "import os\n",
    "import glob\n",
    "import gzip\n",
    "import psycopg2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#USER ENTERED VARIABLES\n",
    "\n",
    "#directory where this file is located and where data files will be stored\n",
    "abs_path = ''\n",
    "\n",
    "#for database \n",
    "host = ''\n",
    "db = ''\n",
    "user = ''\n",
    "password = ''\n",
    "\n",
    "schema_name = 'lodes'\n",
    "\n",
    "#For fresh start (get all files for all states), use state_num = 0 and file_num = 0\n",
    "#Change state_num and file_num if data retrieval needs to be resumed (after SSL Error, etc.)\n",
    "\n",
    "#Enter CURRENT state number\n",
    "#'az' = 0, 'ca' = 1, 'co' = 2, 'dc' = 3, 'fl' = 4, 'il' = 5, 'ma' = 6, 'mi'  = 7, \n",
    "#'mn' = 8, 'nv' = 9, 'ny' = 10, 'tx' = 11, 'wa' = 12\n",
    "state_num = 0\n",
    "\n",
    "#Enter last SUCCESSFUL file number for the current state (For example, in 'az', 71 out of 136, first file will be 72)\n",
    "file_num = 0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#list object containing states to get lodes data from\n",
    "#states = ['az', 'ca', 'co', 'dc', 'fl', 'il', 'ma', 'mi', 'mn', 'nv', 'ny', 'tx', 'wa']\n",
    "states = ['ca']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#FUNCTIONS\n",
    "\n",
    "#Function to get all file names from the associated url \n",
    "#reference: https://stackoverflow.com/questions/11023530/python-to-list-http-files-and-directories\n",
    "def listFD(url, ext=''):\n",
    "    page = requests.get(url).text\n",
    "    #print(page)\n",
    "    soup = BeautifulSoup(page, 'html.parser')\n",
    "    return [url + '/' + node.get('href') for node in soup.find_all('a') if node.get('href').endswith(ext)]\n",
    "\n",
    "#function for psycopg2 to connect to the PostgreSQL database server\n",
    "#reference: https://github.com/NaysanSaran/pandas2postgresql/blob/master/notebooks/CompleteExample.ipynb\n",
    "def connect(params_dic):\n",
    "    \"\"\" Connect to the PostgreSQL database server \"\"\"\n",
    "    conn = None\n",
    "    try:\n",
    "        # connect to the PostgreSQL server\n",
    "        print('Connecting to the PostgreSQL database...')\n",
    "        conn = psycopg2.connect(**params_dic)\n",
    "\n",
    "    except (Exception, psycopg2.DatabaseError) as error:\n",
    "        print(error)\n",
    "        sys.exit(1) \n",
    "    return conn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#CONNECT TO DATABASE\n",
    "\n",
    "#specify the psql connection parameters\n",
    "param_dic = {\n",
    "    \"host\"      : host,\n",
    "    \"database\"  : db,\n",
    "    \"user\"      : user,\n",
    "    \"password\"  : password,\n",
    "}\n",
    "\n",
    "#connect to the database server\n",
    "conn = connect(param_dic)\n",
    "\n",
    "#define the cursor\n",
    "cur = conn.cursor()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#create new tables with san diego subset\n",
    "for y in range(len(states) - state_num):\n",
    "    \n",
    "    #adjust to current state\n",
    "    i = y + state_num\n",
    "    \n",
    "    #list object to contain names of all files to be downloaded from url - for current state\n",
    "    filenames = []\n",
    "    \n",
    "    #variable for url to the current state's lodes files\n",
    "    url = 'https://lehd.ces.census.gov/data/lodes/LODES7/' + states[i] + '/od/'\n",
    "    \n",
    "    #file extension of lodes data\n",
    "    ext = 'csv.gz'\n",
    "\n",
    "    #get all file names for the current state\n",
    "    for file in listFD(url, ext):\n",
    "        #print(file)\n",
    "        filenames.append(os.path.basename(file))\n",
    "    \n",
    "    print('State: ' +states[i] + ', Number of files for ' + url +' = ' + str(len(filenames)))\n",
    "    \n",
    "        \n",
    "    #for each file for the current state:\n",
    "    for z in range(len(filenames) - file_num):\n",
    "            \n",
    "        #adjust to next file\n",
    "        file = z + file_num\n",
    "      \n",
    "        #DATABASE SET UP- SUBSET SAN DIEGO DATA- CREATE NEW TABLES\n",
    "        \n",
    "        #database table name - same as filename, but remove extensions (.csv.gz)\n",
    "        table_name = schema_name + '.' + filenames[file][:-7]\n",
    "       \n",
    "        #database table names for san diego subset\n",
    "        sd_table_name = schema_name + '.' + filenames[file][:-7] + '_sd'\n",
    "        \n",
    "        #drop table if exists\n",
    "        drop_table = ('DROP TABLE IF EXISTS {};'.format(sd_table_name))\n",
    "\n",
    "        try:\n",
    "            cur.execute(drop_table)\n",
    "            print('Table dropped if exists.')\n",
    "        except:\n",
    "            print(\"Error. Table (if exists) not dropped.\")\n",
    "            \n",
    "        #create table\n",
    "        create_table = (\"CREATE TABLE {} AS SELECT * FROM {} WHERE substring(w_geocode from 0 for 6) = '06073' AND substring(h_geocode from 0 for 6) = '06073';\".format(sd_table_name, table_name))\n",
    "\n",
    "        try:\n",
    "            cur.execute(create_table)\n",
    "            print('Table created.')\n",
    "        except:\n",
    "            print(\"Error. Table not created.\")\n",
    "            \n",
    "    #commit database changes for each state\n",
    "    conn.commit()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "#some of the san diego subset tables are empty. Find and delete those tables.\n",
    "#reference: https://pynative.com/python-postgresql-select-data-from-table/\n",
    "#reference: https://dataedo.com/kb/query/postgresql/find-empty-tables-in-database\n",
    "            \n",
    "find_empty_tables = (\"select c.relname as table_name from pg_class c join pg_namespace n on n.oid = c.relnamespace where c.relkind = 'r' and n.nspname not in ('information_schema','pg_catalog') and c.reltuples = 0 order by table_name;\")   \n",
    "#n.nspname as table_schema,       \n",
    "try:\n",
    "    cur.execute(\"set search_path to lodes, public;\")\n",
    "    cur.execute(find_empty_tables)\n",
    "    empty_tables = cur.fetchall() \n",
    "except:\n",
    "    print(\"Error.\") \n",
    "\n",
    "empty_tables_list = []\n",
    "\n",
    "for row in range(len(empty_tables)):\n",
    "    \n",
    "    nested_list = list(empty_tables[row])\n",
    "    split = nested_list[0]\n",
    "    empty_tables_list.append(split)\n",
    "\n",
    "num_empty_tables = len(empty_tables_list)\n",
    "print(num_empty_tables)\n",
    "\n",
    "count = 1\n",
    "\n",
    "#drop empty tables from database\n",
    "for row in range(len(empty_tables_list)):\n",
    "    \n",
    "    table_name = empty_tables_list[row]\n",
    "    print(table_name)\n",
    "    \n",
    "    drop_table = ('DROP TABLE IF EXISTS {};'.format(table_name))\n",
    "\n",
    "    try:\n",
    "        cur.execute(drop_table)\n",
    "        print('Table dropped if exists. #{}'.format(count))\n",
    "    except:\n",
    "        print(\"Error. Table (if exists) not dropped.\")\n",
    "        \n",
    "    count += 1\n",
    "                   \n",
    "#commit database changes for each state\n",
    "conn.commit()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "#delete CA tables, keep only San Diego subset tables.\n",
    "for y in range(len(states) - state_num):\n",
    "    \n",
    "    #adjust to current state\n",
    "    i = y + state_num\n",
    "    \n",
    "    #list object to contain names of all files to be downloaded from url - for current state\n",
    "    filenames = []\n",
    "    \n",
    "    #variable for url to the current state's lodes files\n",
    "    url = 'https://lehd.ces.census.gov/data/lodes/LODES7/' + states[i] + '/od/'\n",
    "    \n",
    "    #file extension of lodes data\n",
    "    ext = 'csv.gz'\n",
    "\n",
    "    #get all file names for the current state\n",
    "    for file in listFD(url, ext):\n",
    "        #print(file)\n",
    "        filenames.append(os.path.basename(file))\n",
    "    \n",
    "    print('State: ' +states[i] + ', Number of files for ' + url +' = ' + str(len(filenames)))\n",
    "    \n",
    "        \n",
    "    #for each file for the current state:\n",
    "    for z in range(len(filenames) - file_num):\n",
    "            \n",
    "        #adjust to next file\n",
    "        file = z + file_num\n",
    "        \n",
    "        #database table name - same as filename, but remove extensions (.csv.gz)\n",
    "        table_name = schema_name + '.' + filenames[file][:-7]\n",
    "        print(table_name)\n",
    "        \n",
    "        #drop table if exists\n",
    "        drop_table = ('DROP TABLE IF EXISTS {};'.format(table_name))\n",
    "\n",
    "        try:\n",
    "            cur.execute(drop_table)\n",
    "            print('Table dropped if exists.')\n",
    "        except:\n",
    "            print(\"Error. Table (if exists) not dropped.\")\n",
    "            \n",
    "#commit changes to the database\n",
    "conn.commit()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "#Aggregate data in SD tables to block groups\n",
    "#Part 1: get table names to aggregate\n",
    "find_lodes_tables =(\"SELECT table_name FROM information_schema.tables WHERE table_schema = 'lodes'\")\n",
    "\n",
    "try:\n",
    "    cur.execute(\"set search_path to lodes, public;\")\n",
    "    cur.execute(find_lodes_tables)\n",
    "    lodes_tables = cur.fetchall() \n",
    "except:\n",
    "    print(\"Error.\") \n",
    "    \n",
    "lodes_tables_list = []\n",
    "nested_list = []\n",
    "\n",
    "for row in range(len(lodes_tables)):\n",
    "    \n",
    "    nested_list = list(lodes_tables[row])\n",
    "    split = nested_list[0]\n",
    "    lodes_tables_list.append(split)\n",
    "    \n",
    "#print(len(lodes_tables_list))\n",
    "#print(lodes_tables_list)\n",
    "\n",
    "#delete shapefiles from list\n",
    "lodes_tables_list = lodes_tables_list[:-3]\n",
    "\n",
    "#print(len(lodes_tables_list))\n",
    "#print(lodes_tables_list)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "#Aggregate data in SD tables to block groups\n",
    "#Part 1: create tables with aggregated data\n",
    "for row in range(len(lodes_tables_list)):\n",
    "    \n",
    "    table_name = schema_name + '.' + lodes_tables_list[row]\n",
    "    \n",
    "    aggregated_table = table_name + '_bg'\n",
    "    #print(aggregated_table)\n",
    "\n",
    "    drop_table = (\"DROP TABLE IF EXISTS {};\".format(aggregated_table))\n",
    "\n",
    "    try:\n",
    "        cur.execute(drop_table)\n",
    "        print('Table dropped if exists.')\n",
    "    except:\n",
    "        print(\"Error. Table (if exists) not dropped.\")\n",
    "\n",
    "    create_table = (\"CREATE TABLE {} AS SELECT t.w_geocode, t.h_geocode, SUM(t.s000) AS s000, SUM(t.sa01) AS sa01, SUM(t.sa02) AS sa02, SUM(t.sa03) AS sa03, SUM(t.se01) AS se01, SUM(t.se02) AS se02, SUM(t.se03) AS se03, SUM(t.si01) AS si01, SUM(t.si02) AS si02, SUM(t.si03) AS si03 FROM (SELECT SUBSTRING(w_geocode from 0 for 13) AS w_geocode, SUBSTRING(h_geocode from 0 for 13) AS h_geocode, s000, sa01, sa02, sa03, se01, se02, se03, si01, si02, si03 FROM {}) AS t GROUP BY t.w_geocode, t.h_geocode;\".format(aggregated_table, table_name))\n",
    "    #create_table = (\"CREATE TABLE {} AS SELECT t.w_geocode, t.h_geocode, SUM(t.s000) AS s000, SUM(t.sa01) AS sa01, SUM(t.sa02) AS sa02, SUM(t.sa03) AS sa03, SUM(t.se01) AS se01, SUM(t.se02) AS se02, SUM(t.se03) AS se03, SUM(t.sl01) AS sl01, SUM(t.sl02) AS sl02, SUM(t.sl03) AS sl03 FROM (SELECT SUBSTRING(w_geocode from 0 for 13) AS w_geocode, SUBSTRING(h_geocode from 0 for 13) AS h_geocode, s000, sa01, sa02, sa03, se01, se02, se03, sl01, sl02, sl03 FROM {}) AS t GROUP BY t.w_geocode, t.h_geocode;\".format(aggregated_table, table_name))\n",
    "\n",
    "    try:\n",
    "        cur.execute(create_table)\n",
    "        print('Table created.')\n",
    "    except:\n",
    "        print(\"Error. Table not created.\")\n",
    "\n",
    "conn.commit()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "#Delete tables with non-aggregated data\n",
    "for row in range(len(lodes_tables_list)):\n",
    "    \n",
    "    table_name = schema_name + '.' + lodes_tables_list[row]\n",
    "\n",
    "    drop_table = (\"DROP TABLE IF EXISTS {};\".format(table_name))\n",
    "\n",
    "    try:\n",
    "        cur.execute(drop_table)\n",
    "        print('Table dropped if exists.')\n",
    "    except:\n",
    "        print(\"Error. Table (if exists) not dropped.\")\n",
    "\n",
    "conn.commit()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#create indexes for aggregated tables\n",
    "for row in range(len(lodes_tables_list)):\n",
    "    \n",
    "    table_name = schema_name + '.' + lodes_tables_list[row] + '_bg'\n",
    "    index_name = 'idx_' + lodes_tables_list[row] + '_bg_wh_geocode'\n",
    "    \n",
    "    create_index = (\"create index {} on {} (w_geocode, h_geocode);\".format(index_name, table_name))\n",
    "    \n",
    "    try:\n",
    "        cur.execute(create_index)\n",
    "        print('Index created.')\n",
    "    except:\n",
    "        print(\"Error. Index not created.\")\n",
    "\n",
    "conn.commit()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#COMMIT AND CLOSE DATABASE\n",
    "#commit changes to the database\n",
    "conn.commit()\n",
    "\n",
    "#close the database connection\n",
    "conn.close()\n",
    "\n",
    "#close the cursor\n",
    "cur.close()"
   ]
  },
  {
   "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.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}