2-data-harmonization.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import datetime\n",
    "import os\n",
    "import sqlite3\n",
    "\n",
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Data Harmonization\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Collected Data\n",
    "\n",
    "# 1. EHR tables in a relational database. Stored in `ehr/ehr.db`\n",
    "patients = ['patient_id', 'name', 'dob', 'gender', 'race', 'postal_code', 'occupation', 'phone']\n",
    "\n",
    "admissions = ['patient_id', 'hadm_id', 'intime', 'outtime']\n",
    "\n",
    "procedures = ['hadm_id', 'icd9_code', 'clinician_id']\n",
    "\n",
    "diagnoses = ['hadm_id', 'icd9_code', 'clinician_id']\n",
    "\n",
    "chartevents = ['hadm_id', 'item_id', 'value', 'datetime']\n",
    "\n",
    "d_procedures = {3000:'Carotid endarterectomy'}\n",
    "d_diagnoses = {10000:'diabetes', 10001:'hiv', 10002:'cancer'}\n",
    "d_chartevents = {2000:'weight', 2001:'abp'}\n",
    "\n",
    "\n",
    "# 2. Streamed blood pressure numberics from wearable device.\n",
    "# Stored in csv files: `blood-pressure/<patient_id>-<date>.csv`.\n",
    "# Aggregated hourly data. Each patients have recordings both before and after their procedure.\n",
    "\n",
    "# example format:\n",
    "\"\"\"\n",
    "hour,bp_min,bp_max\n",
    "0,70,72\n",
    "...\n",
    "23,45,60\n",
    "\"\"\"\n",
    "\n",
    "# 3. Pharmacy orders. Stored in text files: pharmacy/NNNNNN.txt\n",
    "\n",
    "# example format:\n",
    "\"\"\"\n",
    "Date:YYYY-MM-DD\n",
    "Patient ID: \n",
    "Pharmacy: \n",
    "Drug: dexamethasone/erlotinib/other\n",
    "\"\"\"\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Research database tables\n",
    "\n",
    "# Removed name, phone\n",
    "patients = ['patient_id', 'dob', 'gender', 'race', 'postal_code', 'occupation'] \n",
    "\n",
    "admissions = ['patient_id', 'hadm_id', 'intime', 'outtime']\n",
    "\n",
    "# Removed clinician_id\n",
    "procedures = ['hadm_id', 'icd9_code'] \n",
    "\n",
    "# Removed clinician_id\n",
    "diagnoses = ['hadm_id', 'icd9_code']\n",
    "\n",
    "chartevents = ['hadm_id', 'item_id', 'value', 'datetime']\n",
    "\n",
    "# Convert file to rdb table\n",
    "bpstream = ['patient_id', 'datetime', 'bp_min', 'bp_max']\n",
    "\n",
    "# Convert file to rdb table\n",
    "medications = ['patient_id', 'datetime', 'drug']\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Task: Combine the disparate data sources to create a single usable sharable research database"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Step 1 - Get the EHR data\n",
    "\n",
    "# Read the data from the EHR\n",
    "conn_ehr = sqlite3.connect('data/example-data-lake/ehr/ehr.db')\n",
    "\n",
    "df_patients = pd.read_sql_query('select * from patients;', conn_ehr)\n",
    "df_admissions = pd.read_sql_query('select * from admissions;', conn_ehr)\n",
    "df_procedures = pd.read_sql_query('select * from procedures;', conn_ehr)\n",
    "df_diagnoses = pd.read_sql_query('select * from diagnoses;', conn_ehr)\n",
    "df_chartevents = pd.read_sql_query('select * from chartevents;', conn_ehr)\n",
    "\n",
    "conn_ehr.close()\n",
    "\n",
    "# Remove the unwanted columns\n",
    "del(df_patients['name'])\n",
    "del(df_patients['phone'])\n",
    "\n",
    "del(df_procedures['clinician_id'])\n",
    "del(df_diagnoses['clinician_id'])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Step 2 - Get the hourly blood pressure stream\n",
    "file_dir = 'data/example-data-lake/bp-stream'\n",
    "\n",
    "# Holds all the dataframes from the bp files\n",
    "bp_dataframes = []\n",
    "\n",
    "for file in os.listdir(file_dir):\n",
    "    # Read the csv file\n",
    "    df = pd.read_csv(os.path.join(file_dir, file))\n",
    "    # The file name contains information\n",
    "    patient_id, year, month, day = file.strip('.csv').split('-')\n",
    "    \n",
    "    # Add the necessary columns \n",
    "    df['datetime'] = [datetime.datetime(int(year), int(month), int(day), h) for h in df.loc[:, 'hour']]\n",
    "    df['patient_id'] = int(patient_id)\n",
    "    df = df.loc[:, ['patient_id', 'datetime', 'bp_min', 'bp_max']]\n",
    "    \n",
    "    bp_dataframes.append(df)\n",
    "\n",
    "# Concatenate all the streamed bp data\n",
    "df_bpstream = pd.concat(bp_dataframes, ignore_index=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Step 3 - Pharmacy data\n",
    "\n",
    "def read_pharm_file(filename):\n",
    "    \"\"\"\n",
    "    Read a pharmacy claim file. Lines are datetime, patient id, pharmacy, drug.\n",
    "    \n",
    "    Return patient_id, datetime, drug\n",
    "    \"\"\"\n",
    "    with open(filename) as f:\n",
    "        lines = f.readlines()\n",
    "        \n",
    "    lines = [l.strip() for l in lines]\n",
    "        \n",
    "    pid = int(lines[1].split(':')[-1])\n",
    "    year, month, day = lines[0].split(':')[-1].split('-')\n",
    "    dt = datetime.datetime(int(year), int(month), int(day))\n",
    "    drug = lines[3].split(':')[-1]\n",
    "    \n",
    "    return pid, dt, drug\n",
    "        \n",
    "file_dir = 'data/example-data-lake/pharmacy'\n",
    "\n",
    "# Holds all the info from the pharmacyfiles\n",
    "pharm_data = []\n",
    "\n",
    "for file in os.listdir(file_dir):\n",
    "    # Read the file. The filename contains no information.\n",
    "    data = read_pharm_file(os.path.join(file_dir, file))\n",
    "    pharm_data.append(data)\n",
    "    \n",
    "df_medications = pd.DataFrame(pharm_data, columns=['patient_id', 'datetime', 'drug'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Now we have all the tables we need. Write them to the research database.\n",
    "\n",
    "if os.path.isfile('data/allhealth.db'):\n",
    "    os.remove('data/allhealth.db')\n",
    "            \n",
    "conn = sqlite3.connect('data/allhealth.db')\n",
    "c = conn.cursor()\n",
    "\n",
    "# Insert data\n",
    "df_patients.to_sql('patients', conn, index=False)\n",
    "df_admissions.to_sql('admissions', conn, index=False)\n",
    "df_procedures.to_sql('procedures', conn, index=False)\n",
    "df_diagnoses.to_sql('diagnoses', conn, index=False)\n",
    "df_chartevents.to_sql('chartevents', conn, index=False)\n",
    "df_bpstream.to_sql('bpstream', conn, index=False)\n",
    "df_medications.to_sql('medications', conn, index=False)\n",
    "\n",
    "# Save (commit) the changes\n",
    "conn.commit()\n",
    "\n",
    "# We can also close the connection if we are done with it.\n",
    "# Just be sure any changes have been committed or they will be lost.\n",
    "conn.close()\n"
   ]
  },
  {
   "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.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}