<jupyter_start><jupyter_text>mCODE to OMOP Converter PrototypeFilename: mcode2omop.ipyb Author: <NAME> Created: 1/31/22 Last Updated: 4/13/22 Version: 0.6 Notes: Updated script to insert dataframe into OMOP CDM DB<jupyter_code># import modules
import sys
sys.path.append('/Users/mayt/mltApps/py3/pyMitreOps')  # specify the filepath for the MitreOps libraries

import pandas as pd
import json
import yaml
import requests
import lib.mitrehealth_gen_db as mhdb

# set global configuration options
pd.set_option("display.max_columns", None) 
pd.get_option("display.max_columns")<jupyter_output><empty_output><jupyter_text>database connections<jupyter_code># ********** MAIN SCRIPT (with mysql): gather DB connection config info **********
config_filename = "./config/config_omop.yml"
mh_d_config_info, mh_f_config_info = mhdb.config_init(config_filename)
print(f"mh_d_config_info: {mh_d_config_info}")
print(f"mh_f_config_info: {mh_f_config_info}")
# retrieve the database connection handle. This will be used for OpenEMR DB queries
dbcon_handle = mhdb.db_connect(mh_d_config_info)
# close the OpenEMR database connection
mhdb.db_disconnect(dbcon_handle,mh_d_config_info)
# ********** MAIN SCRIPT (with psql): gather DB connection config info **********
config_filename = "./config/config_omop.yml"
mh_d_config_info_psql, mh_f_config_info_psql = mhdb.config_init(config_filename)
print(f"in main - d_config_info: {mh_d_config_info_psql}")
# retrieve the database connection handle. This will be used for OpenEMR DB queries
dbcon_handle_psql = mhdb.db_connect(mh_d_config_info_psql)
print(f"psql connection string: {dbcon_handle_psql}")
# test the database connection. Unlike mysql, postgresql calls disconnect after the call is done.
df_omop_rs = pd.read_sql("""
                 SELECT * FROM omop_vocab.vocabulary ORDER by vocabulary_id
                 """,
                 con = dbcon_handle_psql)
df_omop_rs<jupyter_output><empty_output><jupyter_text>FHIR Exploration: CancerPatient<jupyter_code># assumes we have already retrieved the configuration info called by function mhdb.config_init(config_filename)
print(mh_f_config_info)
fhir_url = mh_f_config_info['fhir_endpoint']
# resource qualifier to make this more modular
fhir_resource_name = "Patient"

fhir_get_resource = fhir_url + "/" + fhir_resource_name

test_url = "https://api.logicahealth.org/mcodestu2/open/Patient/cancer-patient-eve-anyperson"

r = requests.get(url = test_url) # with parameters 

# extracting data in json format
data = r.json()
data
# using fhir.resources 6.2.0 (https://pypi.org/project/fhir.resources/6.2.0/)
from fhir.resources.patient import Patient
# populate the Patient object from the retrieved FHIR resource. fhir.resources will read and verify the resourceType of Patient before populating.
pat = Patient(**data)
print(pat.id)
print(pat.name[0].family)
print(pat.name[0].given[0])
print(pat.meta.source)
print(pat.gender)
print(pat.address[0].city)
print(pat.communication[0].language.coding[0].code)
print(pat.meta.lastUpdated)
birthDate = pat.birthDate
print(type(pat.birthDate))
print(birthDate.year)
# strip the timestamp from the date; for several tables, OMOP dates are required and do not contain time.
# we should make this a convenience utility function.
# put date variable of interest into a temp variable
full_datetime = pat.meta.lastUpdated

from datetime import datetime
shortDate = full_datetime.strftime('%Y-%m-%d')
print(shortDate)
print((pat.meta.lastUpdated).strftime('%Y-%m-%d'))<jupyter_output><empty_output><jupyter_text>**Handling Extensions*** Extensions are handled pythonically. * Needs exception handling for missing extensions.<jupyter_code>type_check = type(pat.extension)
print(type(pat.extension))
print((type_check, isinstance(type_check, type(None)))) 
print(type_check is None)

# check if an extension exists. If so, then we process it.    
if pat.extension:
    print("pat.extension exists")
else:
    print("pat.extension does not exist")
    
# play cell to test handling paths and values.
# if an extension exists, then get the canonical URLs and the value of the 2nd extension list entry.
if pat.extension:
    # print("pat.extension exists")
    print(pat.extension[0].url)
    print(pat.extension[1].url)
    print(pat.extension[1].valueCode)
# get FHIR extension values
# check the URLs with each extension
# print(len(pat.extension))
# initialize variables
race_category_code = 0
ethnicity_category_code = 0
birth_sex_code = 0

if pat.extension:
    i = 0
    while i < (len(pat.extension)):
        print(i)
        temp_url_string = pat.extension[i].url
        print(temp_url_string)
        # simp_code = pat.extension[i].valueCoding.code
        # print(f"code: {complex_code}")
        if temp_url_string == "http://hl7.org/fhir/us/core/StructureDefinition/us-core-race":
            complex_code = pat.extension[i].extension[0].valueCoding.code
            race_category_code = complex_code
            print(f"race_category_code: {race_category_code}")
            i += 1
            continue         
        elif temp_url_string == "http://hl7.org/fhir/us/core/StructureDefinition/us-core-ethnicity":
            ethnicity_category_code == complex_code
            print(f"ethnicity_category_code: {ethnicity_category_code}")
            i += 1
            continue           
        elif temp_url_string == "http://hl7.org/fhir/us/core/StructureDefinition/us-core-birthsex":
            simp_code = pat.extension[i].valueCode
            birth_sex_code = simp_code
            print(f"birth_sex_code: {birth_sex_code}")
            i += 1
            continue          
        else:
            print("url not found")
            continue

# print(pat.extension[0].extension[0].valueCoding.code)
# create an OMOP dataframe for PERSON.
df_omop_person = pd.DataFrame({'person_id': pd.Series(dtype='int'),
                    'gender_concept_id': pd.Series(dtype='int'),
                    'year_of_birth': pd.Series(dtype='int'),
                    'month_of_birth': pd.Series(dtype='int'),
                    'day_of_birth': pd.Series(dtype='int'),
                    'birth_datetime': pd.Series(dtype='str'),
                    'race_concept_id':pd.Series(dtype='int'),
                    'ethnicity_concept_id':pd.Series(dtype='int'),
                    'location_id':pd.Series(dtype='int'),
                    'provider_id':pd.Series(dtype='int'),
                    'care_site_id':pd.Series(dtype='int'),
                    'person_source_value':pd.Series(dtype='str'),
                    'gender_source_value':pd.Series(dtype='str'),
                    'gender_source_concept_id':pd.Series(dtype='int'),
                    'race_source_value':pd.Series(dtype='str'),
                    'race_source_concept_id':pd.Series(dtype='int'),
                    'ethnicity_source_value':pd.Series(dtype='str'),
                    'ethnicity_source_concept_id':pd.Series(dtype='int'),
                    })
df_omop_person
# create an omop person_id. 
# Because person_id is an integer and FHIR ids are a string or uuid, we'll need to create a conversion/mapping table.
# regardless, we should keep one anyway and document that this needs to be maintained by each omop_site (federated or centralized)
# for our prototype, the file pid_map.csv was created for the mapping.
df_pid_map = pd.read_csv ('./xlat_content/pid_map.csv')
df_pid_map
# check if the id already exists in the mapping. 
# If yes, assign OMOP person_id integer. 
# If not, then augment the highest integer value for a new omop_id and add a new line to the map file
# populate omop_person_id into a list to identify the largest integer (max value)

# search for a row that has a matching fhir_id
df_pid_tmp = df_pid_map.loc[(df_pid_map['fhir_pid'] == pat.id)]
df_pid_tmp
# if the dataframe is empty then append a new row with the pid and the highest integer value
if not df_pid_tmp.empty:
    print("match found")
    # assign omop_person_id to a variable to populate person_id in the omop record
    omop_person_id = df_pid_tmp['omop_person_id'].values[0]
    print(f"omop_person_id: {omop_person_id}")
elif df_pid_tmp.empty:
    print("empty return")
    # get the max value in the omop_person_id
    omop_pids = df_pid_map['omop_person_id']
    omop_pid_max_id = omop_pids.max()
    print(omop_pid_max_id)
    new_omop_pid = omop_pid_max_id + 1
    print(new_omop_pid)
    # create a dictionary of the new value
    dict_new_fhir_omop_id_entry = [{'fhir_pid': pat.id, 'omop_person_id': new_omop_pid}]
    # append the new fhir_id and the id to the temp dataframe
    df_pid_map = df_pid_map.append(dict_new_fhir_omop_id_entry, ignore_index=True)
    print(df_pid_map)
    omop_person_id = new_omop_pid
    print(f"omop_person_id: {omop_person_id}")
    # finally, export the dataframe with the new entry to the CSV file
    df_pid_map.to_csv('./xlat_content/pid_map.csv')
else:
    print("unknown")
# Handle birthsex code to enter gender if there is value in Patient.extension.birth_sex. 
# This must be done after the check for extensions logic.
if birth_sex_code == 0:
    birth_sex_code = pat.gender
# try the dataframe append logic with FHIR elements
dict_omop_entry = [{'person_id': omop_person_id,
                    'gender_concept_id': 0, 
                    'year_of_birth': birthDate.year, 
                    'month_of_birth': birthDate.month, 
                    'day_of_birth': birthDate.day,
                    'birth_datetime': pat.birthDate,
                    'race_concept_id': 0, 
                    'ethnicity_concept_id': 0,
                    'person_source_value': pat.id,
                    'race_source_value': race_category_code,
                    'gender_source_value': birth_sex_code, # pat.gender is the alternative
                    'ethnicity_source_value': ethnicity_category_code
                   }]

# convert from dictionary to dataframe
df_omop_entry = pd.DataFrame.from_dict(dict_omop_entry)

# combined data
df_omop_new = pd.concat([df_omop_person,df_omop_entry], ignore_index=True, sort=False)

print("\n  **  Combined Data  **")

df_omop_new
# Insert Dataframe into SQL Server:

df_omop_new.to_sql('person', con=dbcon_handle_psql, schema='omop_cdm_mitre', if_exists='replace',
          index=False)
# test the database connection. Unlike mysql, postgresql calls disconnect after the call is done.
df_omop_rs = pd.read_sql("""
                 SELECT * FROM omop_cdm_mitre.person LIMIT 10
                 """,
                 con = dbcon_handle_psql)

df_omop_rs
# output to CSV
df_omop_new.to_csv('./output/person.csv', index=False)<jupyter_output><empty_output><jupyter_text>Map Name to Observation<jupyter_code># create an OMOP dataframe for Observation
df_omop_observation = pd.DataFrame({'observation_id': pd.Series(dtype='int'),
                    'person_id': pd.Series(dtype='int'),
                    'observation_concept_id': pd.Series(dtype='int'),
                    'observation_date': pd.Series(dtype='str'),
                    'observation_datetime': pd.Series(dtype='str'),
                    'observation_type_concept_id':pd.Series(dtype='int'),
                    'value_as_number':pd.Series(dtype='float'),
                    'value_as_string':pd.Series(dtype='str'),
                    'value_as_concept_id':pd.Series(dtype='int'),
                    'care_site_id':pd.Series(dtype='int'),
                    'qualifier_concept_id':pd.Series(dtype='int'),
                    'unit_concept_id':pd.Series(dtype='str'),
                    'provider_id':pd.Series(dtype='int'),
                    'visit_occurrence_id':pd.Series(dtype='int'),
                    'visit_detail_id':pd.Series(dtype='int'),
                    'unit_source_value':pd.Series(dtype='str'),
                    'observation_source_value':pd.Series(dtype='str'),
                    'unit_source_value':pd.Series(dtype='str'),
                    'value_source_value':pd.Series(dtype='str'),
                    'observation_event_id':pd.Series(dtype='int'),
                    'obs_event_field_concept_id':pd.Series(dtype='int'),                                    
                    })
df_omop_observation
# try the dataframe append logic with FHIR elements
df_omop_observation.drop(df_omop_observation.index, inplace=True) # first reset the dataframe contents (just needed for initialization)

OMOP_CONCEPT_ID_NAME = int(4161172) # set constant for Patient name = 4161172; OMOP treats name as an observation.
OMOP_CONCEPT_ID_OBSERVATION_TYPE_EHR = int(32880)  # set source from an "standard algorithm" (implying FHIR-based)

dict_omop_observation_name = [{'observation_id': 1, 
                    'person_id' : omop_person_id,
                    'observation_concept_id' : OMOP_CONCEPT_ID_NAME,
                    'observation_date': (pat.meta.lastUpdated).strftime('%Y-%m-%d'), # code to the last updated datetime of the FHIR patient instance containing name
                    'observation_type_concept_id': OMOP_CONCEPT_ID_OBSERVATION_TYPE_EHR,
                    'value_as_string' : pat.name[0].given[0] + ' ' + pat.name[0].family,
                   }]

# convert from dictionary to dataframe
df_omop_observation_name = pd.DataFrame.from_dict(dict_omop_observation_name)

df_omop_observation_name
# combined data

#df_omop_observation_new = df_omop_observation.append(dict_omop_observation_name, ignore_index=True, sort=False) # append is now deprecated. Use concat().
df_omop_observation_new = pd.concat([df_omop_observation,df_omop_observation_name], ignore_index=True, sort=False)

print("\n  **  Combined Data  **")

# df_omop_observation_new

list(df_omop_observation_new.columns)
df_omop_observation_new
# Insert Dataframe into SQL Server:

df_omop_observation_new.to_sql('observation', con=dbcon_handle_psql, schema='omop_cdm_mitre', if_exists='replace',
          index=False)
# test the database connection. Unlike mysql, postgresql calls disconnect after the call is done.
df_omop_rs = pd.read_sql("""
                 SELECT * FROM omop_cdm_mitre.observation LIMIT 10
                 """,
                 con = dbcon_handle_psql)

df_omop_rs<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10425471963933503, 'fraction_numerical': 0.005424063116370809, 'mean_word_length': 3.3709975369458127, 'pattern_counts': {'":': 3, '<': 23, '<?xml version=': 0, '>': 22, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '605348', 'license': 'Apache-2.0', 'n_tokens_mistral': 4643, 'n_tokens_neox': 4467, 'n_words': 985, 'path': 'omop_etl/mitre_mcode2omop/mcode2omop.ipynb', 'repo_name': 'mcode/mcode-omop', 'size': '14199'}