import utils
import pandas as pd
from constants import ALL,column_names,NO_UNITS,START_DT,END_DT
import logger
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.pipeline import Pipeline
from fuzzywuzzy import fuzz
import re
import random
import units
import transformers
import dask.dataframe as dd
from extract_transform_load import ETLManager
import matplotlib.pyplot as plt
ITEMID = 'itemid'
SUBINDEX = 'subindex'
HADM_ID = 'hadm_id'
UOM_MAP = {
'#': 'number',
'%': 'percent',
'(CU MM|mm3)': 'mm**3',
'24(hr|hour)s?': 'day',
'Deg\\.? F': 'degF',
'Deg\\.? C': 'degC',
'I.U.': 'IU',
'MEQ': 'mEq',
'MM HG': 'mmHg',
'\\+': 'pos',
'\\-': 'neg',
'cmH20': 'cmH2O',
'gms': 'grams',
'kg x m ': 'kg*m',
'lpm': 'liter/min',
'm2': 'm**2',
'mcghr': 'mcg/hr',
'mcgkghr': 'mcg/kg/hr',
'mcgkgmin': 'mcg/kg/min',
'mcgmin': 'mcg/min',
'mghr': 'mg/hr',
'mgkghr': 'mg/kg/hr',
'mgmin': 'mg/min',
'\?F':'degF',
'\?C':'degC',
'Uhr':'U/hr',
'Umin':'U/min',
'/mic l':'1/uL',
'K/uL':'x10e3/uL'
}
"""
EXPLORING MIMIC-III database
"""
class explorer(object):
def __init__(self, mimic_conn=None):
if mimic_conn is None:
mimic_conn = connect()
columns_to_keep = ['label','abbreviation','itemid','linksto','category','unitname']
self.mimic_conn = mimic_conn
self.df_all_defs = item_defs(self.mimic_conn)[columns_to_keep]
self.df_all_defs.set_index('itemid',inplace=True, drop=True)
def search(self,terms,loinc_code=None):
results = None
for term in terms:
score = self.df_all_defs[['category','label','abbreviation','unitname']].applymap(lambda x: fuzzy_score(str(x),term)).max(axis=1)
if results is None: results = score
else: results = pd.concat([results, score], axis=1).max(axis=1)
results.name = 'score'
return self.df_all_defs.join(results.to_frame()).sort_values('score',ascending=False)
def investigate(self, itemid,upper_limit):
info = self.df_all_defs.loc[itemid]
table = info.loc['linksto']
df = pd.read_sql_query('SELECT * FROM mimiciii.{} WHERE itemid={}'.format(table,itemid),self.mimic_conn)
print df.describe(include='all')
#value
print 'value count:',df.value.count()
value = df.value
if value.count() > 0:
print value.value_counts()/df.value.count()
value.value_counts().plot('bar')
plt.show()
#valuenum
print 'valuenum count:',df.valuenum.count()
valuenum = df.loc[df.valuenum < upper_limit].valuenum
if valuenum.count() > 0:
valuenum.hist()
plt.show()
print 'UOM info:'
print df.valueuom.value_counts()
return df
def fuzzy_score(x,y):
if len(x)==0 or len(y) == 0: return 0
x = x.lower()
y = y.lower()
score = pd.np.mean([
fuzz.partial_ratio(x,y),
fuzz.token_sort_ratio(x,y),
fuzz.ratio(x,y)
])
bonus = 10 if (x in y) or (y in x) else 0
return score + bonus
def add_item_mapping(component,item_ids,item_map_fpath='config/mimic_item_map.csv'):
item_map = pd.read_csv(item_map_fpath)
new_mappings = pd.DataFrame([(component,item_id) for item_id in item_ids],columns=['component','itemid'])
item_map = pd.concat([item_map,new_mappings]).reset_index(drop=True)
item_map.to_csv(item_map_fpath, index=False)
return item_map
"""
EXTRACTING Data from MIMIC-III
- Timeseries data
- Context/demographic data
"""
class MimicETLManager(ETLManager):
def __init__(self,hdf5_fname,mimic_item_map_fname,data_dict):
self.conn = connect()
self.item_map_fname = mimic_item_map_fname
self.data_dict = data_dict
cleaners = standard_cleaners(data_dict)
super(MimicETLManager,self).__init__(cleaners,hdf5_fname)
def extract(self,component):
item_map = pd.read_csv(self.item_map_fname)
return extract_component(self.conn,component,item_map)
def transform(self,df,component):
transformers = transform_pipeline(component,self.data_dict)
return transformers.fit_transform(df)
def extracted_ids(self,df_extracted):
return df_extracted[column_names.ID].unique().tolist()
def extracted_data_count(self,df_extracted):
return df_extracted[column_names.VALUE].count()
def all_ids(self):
return get_all_hadm_ids(self.conn)
def extract_component(mimic_conn,component,item_map,hadm_ids=ALL):
itemids = items_for_components(item_map,[component])
if len(itemids) == 0: return None
#Get item defs and filter to what we want
df_item_defs = item_defs(mimic_conn)
df_item_defs = df_item_defs[df_item_defs.itemid.isin(itemids)]
df_item_defs = df_item_defs[~(df_item_defs.linksto == '')]
grouped = df_item_defs.groupby('linksto')
df_list = []
df_columns = column_map()
too_many_ids = len(hadm_ids) > 2000
for table,group in grouped:
itemids = group.itemid.astype(int).tolist()
logger.log('Extracting {} items from {}'.format(len(itemids),table))
is_iemv = table == 'inputevents_mv'
df_col = df_columns.columns.tolist() + (['statusdescription'] if is_iemv else [])
for ix,column_set in df_columns.loc[[table]].iterrows():
psql_col = column_set.tolist() + (['statusdescription'] if is_iemv else [])
query = 'SELECT {} FROM mimiciii.{} WHERE itemid = ANY (ARRAY{})'.format(','.join(psql_col),table,itemids)
if not (hadm_ids == ALL) and not too_many_ids:
query += ' AND hadm_id = ANY (ARRAY{})'.format(hadm_ids)
df = pd.read_sql_query(query,mimic_conn)
df.columns = df_col
if too_many_ids:
df = df[df[column_names.ID].isin(hadm_ids)]
if is_iemv:
df = df.loc[df['statusdescription'].astype(str) != 'Rewritten']
df.drop('statusdescription', axis=1,inplace=True)
df_list.append(df)
logger.log('Combine DF')
df_all = pd.concat(df_list)
return df_all
def get_context_data(hadm_ids=ALL,mimic_conn=None):
if mimic_conn is None:
mimic_conn = connect()
#get HADM info (includes patient demographics)
df_hadm = hadm_data(mimic_conn,hadm_ids)
#get icu data
df_icu = icu_data(mimic_conn,hadm_ids)
#merge into single dataframe
df_hadm_info = df_hadm.merge(df_icu,on=HADM_ID,how='left')
df_hadm_info.rename(columns={HADM_ID : column_names.ID},inplace=True)
return df_hadm_info
def hadm_data(mimic_conn,hadm_ids):
"""
expects a TUPLE of hadm_ids
"""
"""
@@@@@@@@@@@@
1. Get all demographic data from the ADMISSIONS table = df_hadm
https://mimic.physionet.org/mimictables/admissions/
SELECT subject_id, hadm_id, admittime, dischtime, language, religion,
marital_status, ethnicity, diagnosis, admission_location
FROM admissions
WHERE hadm_id IN hadm_ids
@@@@@@@@@@@@
"""
table = 'mimiciii.admissions'
hadm_where_case = None if hadm_ids == ALL else 'hadm_id IN {}'.format(tuple(hadm_ids))
col_psql = ['subject_id', HADM_ID, 'admittime', 'dischtime', 'language',
'religion','marital_status', 'ethnicity', 'diagnosis','admission_location']
col_df = ['pt_id',HADM_ID,START_DT,END_DT,'lang',
'religion','marital_status','ethnicity','dx_info','admission_location']
df_hadm = context_extraction_helper(mimic_conn,table,col_psql,col_df,hadm_where_case)
"""
@@@@@@@@@@@@
2. Get all demographic data from PATIENTS table = df_pt
https://mimic.physionet.org/mimictables/patients/
SELECT gender, dob, dod
FROM patients
WHERE subject_id IN pt_ids
@@@@@@@@@@@@
"""
table = 'mimiciii.patients'
pt_ids = df_hadm['pt_id'].unique().tolist()
col_psql = ['subject_id','gender','dob','dod']
col_df = ['pt_id','gender','dob','dod']
df_pt = context_extraction_helper(mimic_conn,table,col_psql,col_df)
"""
@@@@@@@@@@@@
3. Get all ICD codes data from DIAGNOSES_ICD table = df_icd
https://mimic.physionet.org/mimictables/diagnoses_icd/
SELECT subject_id, hadm_id, seq_num, icd9_code
FROM diagnoses_icd
WHERE hadm_id IN hadm_ids
@@@@@@@@@@@@
"""
table = 'mimiciii.diagnoses_icd'
col_psql = ['subject_id',HADM_ID,'seq_num','icd9_code']
col_df = ['pt_id',HADM_ID,'icd_rank','icd_code']
df_icd = context_extraction_helper(mimic_conn,table,col_psql,col_df,hadm_where_case)
"""
@@@@@@@@@@@@
4. Make df_icd into single rows for each admission, where one
column is an ordered list of ICD codes for that admission
@@@@@@@@@@@@
"""
df_icd = df_icd.sort_values('icd_rank').groupby(HADM_ID).apply(lambda grp: grp['icd_code'].tolist())
df_icd.name = 'icd_codes'
df_icd = df_icd.reset_index()
"""
@@@@@@@@@@@@
Merging
5. Merge df_pt and df_hadm on subject_id = demographics_df
6. Merge demographics_df with df_icd on hadm_id = df_hadm_info
@@@@@@@@@@@@
"""
df_demographics = df_hadm.merge(df_pt,on='pt_id',how='left')
df_hadm_info = df_demographics.merge(df_icd,on=HADM_ID,how='left')
"""
@@@@@@@@@@@@
Cleaning
7. Remove all NA hadm_ids
8. Add age column
9. cast hadm_id to int
@@@@@@@@@@@@
"""
df_hadm_info = df_hadm_info.dropna(subset=[HADM_ID])
df_hadm_info['age'] = df_hadm_info['start_dt']-df_hadm_info['dob']
df_hadm_info[HADM_ID] = df_hadm_info[HADM_ID].astype(int)
return df_hadm_info
def icu_data(mimic_conn,hadm_ids):
table = 'mimiciii.icustays'
col_psql = [HADM_ID,'icustay_id','dbsource','first_careunit','last_careunit','intime','outtime','los']
col_df = [HADM_ID,'icustay_id','dbsource','first_icu','last_icu','intime','outtime','los']
hadm_where_case = None if hadm_ids == ALL else 'hadm_id IN {}'.format(tuple(hadm_ids))
df_icu = context_extraction_helper(mimic_conn,table,col_psql,col_df,hadm_where_case)
"""
Cleaning
- drop ICUSTAYS without hadm_id
"""
df_icu = df_icu.dropna(subset=[HADM_ID])
return df_icu
def context_extraction_helper(mimic_conn,table,col_psql,col_df,where_case=None):
query = utils.simple_sql_query(table,col_psql,where_case)
df = pd.read_sql_query(query,mimic_conn)
rename_dict = dict(zip(col_psql,col_df))
df.rename(index=str,columns=rename_dict,inplace=True)
return df
def column_map():
"""
Create column mapping
"""
#definitions
std_columns = [column_names.ID,column_names.DATETIME,column_names.VALUE,column_names.UNITS,'itemid']
psql_col = ['hadm_id','charttime','value','valueuom','itemid']
col_series = pd.Series(
psql_col,
index=std_columns
)
map_list = []
col_series.name = 'chartevents'
map_list.append(col_series)
col_series = col_series.copy()
col_series.name = 'labevents'
map_list.append(col_series)
col_series = col_series.copy()
col_series.name = 'procedureevents_mv'
map_list.append(col_series)
col_series = col_series.copy()
col_series.name = 'datetimeevents'
map_list.append(col_series)
col_series = col_series.copy()
col_series.name = 'outputevents'
map_list.append(col_series)
psql_col = ['hadm_id','starttime','rate','rateuom','itemid']
col_series = pd.Series(
psql_col,
index=std_columns
)
col_series.name = 'inputevents_mv'
map_list.append(col_series)
psql_col = ['hadm_id','endtime','amount','amountuom','itemid']
col_series = pd.Series(
psql_col,
index=std_columns
)
col_series.name = 'inputevents_mv'
map_list.append(col_series)
psql_col = ['hadm_id','charttime','rate','rateuom','itemid']
col_series = pd.Series(
psql_col,
index=std_columns
)
col_series.name = 'inputevents_cv'
map_list.append(col_series)
psql_col = ['hadm_id','charttime','amount','amountuom','itemid']
col_series = pd.Series(
psql_col,
index=std_columns
)
col_series.name = 'inputevents_cv'
map_list.append(col_series)
return pd.DataFrame(map_list)
def item_defs(mimic_conn):
df_items = pd.read_sql_query('SELECT * FROM mimiciii.d_items',mimic_conn)
df_labitems = pd.read_sql_query('SELECT * FROM mimiciii.d_labitems',mimic_conn)
df_labitems['linksto'] = 'labevents'
df_all_items = pd.concat([df_labitems,df_items])
return df_all_items
def items_for_components(item_map,components=ALL):
if not (components == ALL):
item_map = item_map[item_map.component.isin(components)]
items = item_map.itemid.unique().astype(int).tolist()
return items
def get_all_hadm_ids(conn=None):
if conn is None:
conn = connect()
all_ids = pd.read_sql_query('SELECT hadm_id from mimiciii.admissions',conn)['hadm_id']
all_ids = all_ids[~pd.isnull(all_ids)]
return all_ids.astype(int).sort_values().tolist()
def sample_hadm_ids(n,seed):
all_ids = get_all_hadm_ids()
random.seed(seed)
sampled_ids = random.sample(all_ids,n)
return sampled_ids
def connect(psql_username='postgres',psql_pass='123'):
return utils.psql_connect(psql_username,psql_pass,'mimic')
"""
TRANSFORM Data extracted from MIMIC-III
"""
class CleanUnits(BaseEstimator,TransformerMixin):
def __init__(self,component,data_dict):
self.data_dict = data_dict
self.component = component
def fit(self, x, y=None):
return self
def transform(self, df):
logger.log('Clean UOM',new_level=True)
df = clean_uom(df,self.component,self.data_dict)
logger.end_log_level()
return df
def clean_uom(df,component,data_dict):
grouped = df.groupby(column_names.UNITS)
for old_uom,group in grouped:
new_uom = process_uom(old_uom,component,data_dict)
df.loc[group.index,column_names.UNITS] = new_uom
if not (old_uom == new_uom):
df.loc[group.index,ITEMID] = utils.append_to_description(df.loc[group.index,ITEMID].astype(str),old_uom)
return df
def process_uom(units,component,data_dict):
if units in ['BPM','bpm']:
if component == data_dict.components.HEART_RATE: units = 'beats/min'
if component == data_dict.components.RESPIRATORY_RATE: units = 'breaths/min'
for to_replace,replacement in UOM_MAP.iteritems():
units = re.sub(to_replace, replacement,units,flags=re.IGNORECASE)
return units
class clean_extract(BaseEstimator,TransformerMixin):
def fit(self, x, y=None):
return self
def transform(self, df):
"""
FORMAT pre-unstack columns
"""
df = df.replace(to_replace='', value=pd.np.nan)
#drop NAN record_id, timestamps, or value
df.dropna(subset=[column_names.ID,column_names.DATETIME,column_names.VALUE], how='any',inplace=True)
#ID to integer
df.loc[:,column_names.ID] = df.loc[:,column_names.ID].astype(int)
#DATETIME to pd.DATETIME
df.loc[:,column_names.DATETIME] = pd.to_datetime(df.loc[:,column_names.DATETIME],errors='raise')
#set UOM to NO_UOM if not declared
df.loc[:,column_names.UNITS] = df.loc[:,column_names.UNITS].fillna(NO_UNITS)
df.rename(index=str,columns={ITEMID:column_names.DESCRIPTION},inplace=True)
index_cols = [
column_names.ID,
column_names.DATETIME,
column_names.DESCRIPTION,
column_names.UNITS
]
#Set up our row index
df.set_index(index_cols,inplace=True)
return df
class unstacker(transformers.safe_unstacker):
def __init__(self):
super(unstacker,self).__init__(column_names.UNITS,column_names.DESCRIPTION)
def transform_pipeline(component,data_dict):
return Pipeline([
('clean_units',CleanUnits(component,data_dict)),
('clean_df',clean_extract()),
('unstack',unstacker()),
('add_level',transformers.add_level(component,'component',axis=1)),
])
def standard_cleaners(data_dict):
category_map = mimic_category_map(data_dict)
ureg = units.MedicalUreg()
return Pipeline([
('aggregate_same_datetime',transformers.same_index_aggregator(lambda grp:grp.iloc[0])),
('split_dtype',transformers.split_dtype()),
('standardize_columns',transformers.column_standardizer(data_dict,ureg)),
('standardize_categories',transformers.standardize_categories(data_dict,category_map)),
('split_bad_categories',transformers.split_bad_categories(data_dict)),
('one_hotter',transformers.nominal_to_onehot()),
('drop_oob_values',transformers.oob_value_remover(data_dict))
])
def mimic_category_map(data_dict):
return {
data_dict.components.GLASGOW_COMA_SCALE_EYE_OPENING: {
'1 No Response': 6,
'2 To pain': 7,
'3 To speech': 8,
'4 Spontaneously': 9
},
data_dict.components.GLASGOW_COMA_SCALE_MOTOR: {
'1 No Response': 0,
'2 Abnorm extensn': 1,
'3 Abnorm flexion': 2,
'4 Flex-withdraws': 3,
'5 Localizes Pain': 4,
'6 Obeys Commands': 5
},
data_dict.components.GLASGOW_COMA_SCALE_VERBAL: {
'1 No Response': 10,
'1.0 ET/Trach': 10,
'2 Incomp sounds': 11,
'3 Inapprop words': 12,
'4 Confused': 13,
'5 Oriented':14
}
}
def ETL(extractor,
components,
data_dict,
same_dt_aggregator,
hdf5_fname=None,joined_path=None,
hadm_ids=ALL,
use_base_df=True,
to_pandas=False,
chunksize=500000):
logger.log('***ETL***',new_level=True)
logger.log('SETUP',new_level=True)
category_map = mimic_category_map(data_dict)
ureg = units.MedicalUreg()
transformer = transform_pipeline()
standard_clean_pipeline = Pipeline([
('aggregate_same_datetime',same_dt_aggregator),
('split_dtype',transformers.split_dtype()),
('standardize_columns',transformers.column_standardizer(data_dict,ureg)),
('standardize_categories',transformers.standardize_categories(data_dict,category_map)),
('split_bad_categories',transformers.split_bad_categories(data_dict)),
# ('one_hotter',transformers.nominal_to_onehot()),
('drop_oob_values',transformers.oob_value_remover(data_dict))
])
should_save = (hdf5_fname is not None)
df_base = None
if should_save & use_base_df:
try:
df_base = utils.open_df(hdf5_fname,joined_path)
except:
pass
if df_base is not None:
existing_components = df_base.columns.get_level_values(column_names.COMPONENT).unique().tolist()
existing_ids = set(df_base.index.get_level_values(column_names.ID).tolist())
requested_ids = hadm_ids if hadm_ids != ALL else get_all_hadm_ids()
new_ids = [ID for ID in requested_ids if ID not in existing_ids]
#case 1: new ids in existing columns, don't try to be smart with ALL unless not a lot of IDs
if len(new_ids) > 0:
df_addition = ETL(extractor,
existing_components,
data_dict,
same_dt_aggregator,
hadm_ids=new_ids,
to_pandas=True)
if df_addition is not None:
df_base = pd.concat([df_base,df_addition])
#now we only need to load NEW components
components = [comp for comp in components if comp not in existing_components]
logger.log('Base DF to Dask')
df_base = dd.from_pandas(df_base.reset_index(), chunksize=chunksize)
df_all = df_base
logger.log('BEGIN ETL for {} admissions and {} components: {}'.format(hadm_ids if hadm_ids == ALL else len(hadm_ids),
len(components),
components),new_level=True,end_level=True)
for component in components:
logger.log('{}: {}/{}'.format(component.upper(),components.index(component)+1,len(components)),new_level=True)
"""
@@@@@@@@@@@@@@@
----EXTRACT----
@@@@@@@@@@@@@@@
"""
logger.log("Extracting...",new_level=True)
df_extracted = extractor.extract_component(component,hadm_ids)
if df_extracted.empty:
print 'EMPTY Dataframe EXTRACTED for {}, n={} ids'.format(component,len(hadm_ids))
logger.end_log_level()
continue
if should_save:
logger.log('Save EXTRACTED DF = {}'.format(df_extracted.shape))
utils.save_df(df_extracted,hdf5_fname,'extracted/{}'.format(component))
logger.end_log_level()
"""
@@@@@@@@@@@@@@@@@
----TRANSFORM----
@@@@@@@@@@@@@@@@@
"""
logger.log("Transforming... {}".format(df_extracted.shape),new_level=True)
transformer.set_params(add_level__level_val=component)
df_transformed = transformer.transform(df_extracted)
print 'Data Loss (Extract > Transformed):',utils.data_loss(df_extracted.set_index(column_names.ID).value.to_frame(),df_transformed)
if df_transformed.empty:
print 'EMPTY Dataframe TRANSFORMED for {}, n={} ids'.format(component,len(hadm_ids))
logger.end_log_level()
continue
if should_save:
logger.log('Save TRANSFORMED DF = {}'.format(df_transformed.shape))
utils.save_df(df_transformed,hdf5_fname,'transformed/{}'.format(component))
logger.end_log_level()
"""
@@@@@@@@@@@@@@@
-----CLEAN-----
@@@@@@@@@@@@@@@
"""
logger.log("Cleaning... {}".format(df_transformed.shape),new_level=True)
df = standard_clean_pipeline.transform(df_transformed)
print 'Data Loss (Extract > Cleaned):', utils.data_loss(df_extracted.set_index(column_names.ID).value.to_frame(),df)
if df.empty:
print 'EMPTY Dataframe TRANSFORMED for {}, n={} ids'.format(component,len(hadm_ids))
logger.end_log_level()
continue
if should_save:
logger.log('Save CLEANED DF = {}'.format(df.shape))
utils.save_df(df,hdf5_fname,'cleaned/{}'.format(component))
logger.end_log_level()
del df_extracted,df_transformed
logger.log('Filter & sort - {}'.format(df.shape))
df.sort_index(inplace=True)
df.sort_index(inplace=True, axis=1)
logger.log('Convert to dask - {}'.format(df.shape))
df_dask = dd.from_pandas(df.reset_index(), chunksize=chunksize)
del df
logger.log('Join to big DF')
if df_all is None: df_all = df_dask
else :
df_all = df_all.merge(df_dask,how='outer', on=['id','datetime'])
del df_dask
logger.end_log_level()
logger.end_log_level()
if df_all is None or not to_pandas:
logger.end_log_level()
return df_all
logger.log('Dask DF back to pandas')
df_pd = df_all.compute()
del df_all
df_pd.set_index(['id','datetime'], inplace=True)
logger.log('SORT Joined DF')
df_pd.sort_index(inplace=True)
df_pd.sort_index(inplace=True, axis=1)
if should_save:
logger.log('SAVE Big DF')
utils.save_df(df_pd,hdf5_fname,joined_path)
logger.end_log_level()
return df_pd
