Code fragments of main.py

• import os
• import random
• import sys
• import fhirpy
• import numpy as np
• import pandas as pd
• from matplotlib import pyplot as plt
• from sklearn.preprocessing import LabelEncoder, MinMaxScaler, StandardScaler
• from sklearn.model_selection import train_test_split
• from sklearn.metrics import roc_curve, auc
• from sklearn.svm import SVC
• import tensorflow as tf
• def get_value_from_resource(resource_class: fhirpy.SyncFHIRClient, resource: str):
• """ This function get the value from a FHIR server resource :param resource: you want :param resource_class: a fhir resource where you want to extract the data :return: a str or int from the server """
• try:
• resource_value = resource_class[resource]
• except: resource_value = None
• return resource_value
• def get_value_from_resource_coding(resource_class: fhirpy.SyncFHIRClient, resource1, id):
• """ This function get the value from a FHIR server resource :param resource2: 2 you want to extract :param resource1: 1 you want extract :param resource_class: a fhir resource where you want to extract the data :return: a str or int from the server """
• try:
• resource_value = resource_class['category'][id]['coding'][id][str(resource1)]
• except: resource_value = None
• return resource_value
• def get_value_from_valuequantity(resource_class: fhirpy.SyncFHIRClient, resource1):
• """ This function get the value from a FHIR server resource :param resource1: 1 you want extract :param resource_class: a fhir resource where you want to extract the data :return: a str or int from the server """
• try:
• resource_value = resource_class['valueQuantity'][str(resource1)]
• except: resource_value = None
• return resource_value
• random.seed(744)
• image_path = "Image"
• model_path = "Model"
• if not os.path.exists(image_path):
• os.mkdir(image_path)
• if not os.path.exists(model_path):
• os.mkdir(model_path)
• if os.path.exists("Model/model"):
• model_update = True
• print("MODEL FOUND, LOAD MODEL")
• else:
• model_update = False
• if 'SERVERADDRESS' in os.environ:
• server_url = str(os.environ['SERVERADDRESS'])
• else:
• server_url = "http://localhost:8080/fhir"
• print(server_url)
• print("START READING DATA")
• client = fhirpy.SyncFHIRClient(server_url)
• patients = client.resources("Patient")
• id_list = []
• gender_list = []
• birthDate_list = []
• for patient in patients:
• patient_id = "Patient/" + get_value_from_resource(patient, 'id')
• patient_birthDate = get_value_from_resource(patient, 'birthdate')
• patient_gender = get_value_from_resource(patient, 'gender')
• id_list.append(patient_id)
• birthDate_list.append(patient_birthDate)
• gender_list.append(patient_gender)
• patient_df = pd.DataFrame(list(zip(id_list, birthDate_list, gender_list)), columns=["ID", "Age", "Sex"])
• observations = client.resources("Observation")
• observation_id = []
• observation_loinc = []
• observation_value = []
• observation_patient_id = []
• for observation in observations:
• observation_id.append(get_value_from_resource(observation, 'id'))
• observation_loinc.append(get_value_from_resource_coding(observation, 'code', 0))
• observation_value.append(get_value_from_valuequantity(observation, 'value'))
• observation_patient_id.append(observation['subject']['reference'])
• observation_df = pd.DataFrame(list(zip( observation_patient_id, observation_loinc, observation_value)), columns=["ID", "LOINC", "Value"])
• observation_df = observation_df.pivot(columns="LOINC", values="Value", index="ID").reset_index()
• conditions = client.resources("Condition")
• condition_id = []
• condition_value = []
• for condition in conditions:
• condition_id.append(condition['subject']['reference'])
• condition_value.append(int(condition['code']['coding'][0]['code']))
• condition_df = pd.DataFrame(list(zip( condition_id, condition_value)), columns=["ID", "HeartDisease"])
• data = pd.merge(pd.merge(patient_df, observation_df, on="ID"), condition_df, on="ID")
• data = data.drop('ID',axis=1)
• print("START PREPROCESSING")
• le = LabelEncoder()
• df1 = data.copy(deep=True)
• df1['Sex'] = le.fit_transform(df1['Sex'])
• mms = MinMaxScaler()
• # Normalization
• ss = StandardScaler()
• # Standardization
• df1['Oldpeak'] = mms.fit_transform(df1[['Oldpeak']])
• df1['Age'] = ss.fit_transform(df1[['Age']])
• df1['RestingBP'] = ss.fit_transform(df1[['RestingBP']])
• df1['Cholesterol'] = ss.fit_transform(df1[['Cholesterol']])
• df1['MaxHR'] = ss.fit_transform(df1[['MaxHR']])
• features = df1[df1.columns.drop(['HeartDisease', 'RestingBP', 'RestingECG'])].values
• target = df1['HeartDisease'].values
• x_train, x_test, y_train, y_test = train_test_split(features, target, test_size=0.20, random_state=2)
• x_train, x_val, y_train, y_val = train_test_split(x_train, y_train, test_size=0.20, random_state=2)
• # modeling
• print("START THE MODELLING")
• if not model_update: # Specify the model's architecture
• model = tf.keras.models.Sequential([ tf.keras.layers.Flatten(input_shape=[9]), tf.keras.layers.Dense(100, activation='relu'), tf.keras.layers.Dense(1, activation='sigmoid'), ])
• else:
• model = tf.keras.models.load_model('Model/model')
• # Specify the loss function, optimizer, metrics
• model.compile( loss=tf.keras.losses.binary_crossentropy, optimizer=tf.keras.optimizers.Adam(), metrics=[tf.keras.metrics.binary_accuracy] )
• # Train the model
• history = model.fit( x_train, y_train, epochs=200, validation_data=(x_val, y_val), verbose=0 )
• print("SAVE THE MODEL")
• model.save(os.path.join(model_path, "model"))
• print("SAVE THE PLOTS")
• pd.DataFrame(history.history).plot()
• plt.xlabel("epoch")
• plt.legend(bbox_to_anchor=(0.3, 0.9), loc=2)
• plt.savefig(os.path.join(image_path, "Learning_graph.png"))
• plt.close()
• classifier_svc = SVC(kernel='linear', C=0.1)
• classifier_svc.fit(x_train, y_train)
• y_prediction_keras = model.predict(x_test).ravel()
• fpr_keras, tpr_keras, thresholds_keras = roc_curve(y_test, y_prediction_keras)
• auc_keras = auc(fpr_keras, tpr_keras)
• plt.plot([0, 1], [0, 1], 'k--')
• plt.plot(fpr_keras, tpr_keras, label='Keras (area = {:.3f})'.format(auc_keras))
• plt.xlabel('False positive rate')
• plt.ylabel('True positive rate')
• plt.title('ROC curve')
• plt.legend(loc='best')
• plt.savefig(os.path.join(image_path, "ROC_curve.png"))

Graph