--- a +++ b/utils.py @@ -0,0 +1,204 @@ +import pandas as pd +import numpy as np +from sklearn import metrics +from sklearn.metrics import confusion_matrix, classification_report, average_precision_score, precision_recall_curve, accuracy_score, confusion_matrix, average_precision_score + + + +def episode_metrics(model, x, parameters, rhythm, out_message=False): + """ arg + + """ + # Rhythmn predictions + predictions_qa, predictions_r = model.predict(x) + # If the estimated prob of AF(column 1) is higher than the estimated prob of non-AF(column 0) + # than consider the window as an AF window. + y_predictions = np.argmax(predictions_r, axis=1) + y_truth = np.argmax(rhythm, axis=1) + # Confusion_matrix + cf = confusion_matrix(y_truth, y_predictions, labels=[0, 1]) + TN, FP, FN, TP = cf.ravel() + support = TN+FP+FN+TP + # Sensitivity, recall, or true positive rate + TPR = TP/(TP+FN) + # Specificity or true negative rate + TNR = TN/(TN+FP) + # Precision or positive predictive value + PPV = TP/(TP+FP) + # Negative predictive value + NPV = TN/(TN+FN) + # Fall out or false positive rate + FPR = FP/(FP+TN) + # False negative rate + FNR = FN/(TP+FN) + # F1 score + f1 = metrics.f1_score(y_truth, y_predictions, average=None) + # selecting f1score for positive case if present + f1_pos = f1[1] if len(f1) > 1 else f1[0] + # Area under precision recall curve + #auprc = metrics.average_precision_score(y_truth, rhythm[:,1]) + if out_message: + print(pd.DataFrame(cf).rename(columns={0: "Predicted Non-AF", 1:" Predicted AF"}, index={0: "True Non-AF", 1:"True AF"})) + print("Sensitivity/Recall: %0.4f" % TPR) + print("Specificity: %0.4f" % TNR) + print("Precision/PPV: %0.4f" % PPV) + print("Negative predictive value/NPV: %0.2f" % NPV) + print("False positive rate: %0.4f" % FPR) + print("False negative rate: %0.4f" % FNR) + print("F1 score: %0.4f" % f1_pos) + print('support: ', support) + print('\n\n\n') + episode_metrics = [TPR, TNR, PPV, NPV, FPR, FNR, f1_pos, support] + return episode_metrics + + +def episode_metrics_singletask(model, x, parameters, rhythm, out_message=False): + """ arg + + """ + # Rhythmn predictions + predictions_r = model.predict(x) + # If the estimated prob of AF(column 1) is higher than the estimated prob of non-AF(column 0) + # than consider the window as an AF window. + y_predictions = np.argmax(predictions_r, axis=1) + y_truth = np.argmax(rhythm, axis=1) + # Confusion_matrix + cf = confusion_matrix(y_truth, y_predictions, labels=[0, 1]) + TN, FP, FN, TP = cf.ravel() + support = TN+FP+FN+TP + # Sensitivity, recall, or true positive rate + TPR = TP/(TP+FN) + # Specificity or true negative rate + TNR = TN/(TN+FP) + # Precision or positive predictive value + PPV = TP/(TP+FP) + # Negative predictive value + NPV = TN/(TN+FN) + # Fall out or false positive rate + FPR = FP/(FP+TN) + # False negative rate + FNR = FN/(TP+FN) + # F1 score + f1 = metrics.f1_score(y_truth, y_predictions, average=None) + # selecting f1score for positive case if present + f1_pos = f1[1] if len(f1) > 1 else f1[0] + # Area under precision recall curve + #auprc = metrics.average_precision_score(y_truth, rhythm[:,1]) + if out_message: + print(pd.DataFrame(cf).rename(columns={0: "Predicted Non-AF", 1:" Predicted AF"}, index={0: "True Non-AF", 1:"True AF"})) + print("Sensitivity/Recall: %0.4f" % TPR) + print("Specificity: %0.4f" % TNR) + print("Precision/PPV: %0.4f" % PPV) + print("Negative predictive value/NPV: %0.2f" % NPV) + print("False positive rate: %0.4f" % FPR) + print("False negative rate: %0.4f" % FNR) + print("F1 score: %0.4f" % f1_pos) + print('support: ', support) + print('\n\n\n') + episode_metrics = [TPR, TNR, PPV, NPV, FPR, FNR, f1_pos, support] + return episode_metrics + + +def collecting_individual_metrics(model, x, parameters, rhythm, out_message=False): + """ + """ + individual_metrics = {} + for i in np.unique(parameters['ID']): + # Sub-selecting individuals + p_indx = np.where(parameters['ID'] == i)[0] + x_pID = x[p_indx] + parameters_pID = parameters.iloc[p_indx] + rhythm_pID = rhythm[p_indx] + # Rhythmn predictions + predictions_qa, predictions_r = model.predict(x_pID) + # If the estimated prob of AF(column 1) is higher than the estimated prob of non-AF(column 0) + # than consider the window as an AF window. + y_predictions = np.argmax(predictions_r, axis=1) + y_truth = np.argmax(rhythm_pID, axis=1) + # Confusion_matrix + cf = confusion_matrix(y_truth, y_predictions, labels=[0, 1]) + TN, FP, FN, TP = cf.ravel() + support = TN+FP+FN+TP + # Sensitivity, recall, or true positive rate + TPR = TP/(TP+FN) + # Specificity or true negative rate + TNR = TN/(TN+FP) + # Fall out or false positive rate + FPR = FP/(FP+TN) + # False negative rate + FNR = FN/(TP+FN) + # F1 score + f1 = metrics.f1_score(y_truth, y_predictions, average=None) + # selecting f1score for positive case if present + f1_pos = f1[1] if len(f1) > 1 else f1[0] + # Area under precision recall curve + #auprc = metrics.average_precision_score(y_test_truth, rhythm_pID[:,1]) + if (TP + FN) > 0: + individual_metrics[i] = [TPR, np.NaN, np.NaN, FNR, f1_pos, support] + if (FP + TN) > 0: + individual_metrics[i] = [np.NaN, TNR, FPR, np.NaN, np.NaN, support] + if out_message: + print("PATIENT :", i , '\n') + print(pd.DataFrame(cf).rename(columns={0: "Predicted Non-AF", 1:" Predicted AF"}, index={0: "True Non-AF", 1:"True AF"})) + if (TP + FN) > 0: + print("Sensitivity/Recall: %0.4f" % TPR) + print("False negative rate: %0.4f" % FNR) + print("F1 score: %0.4f" % f1_pos) + print('support: ' , support) + if (FP + TN) > 0: + print("Specificity: %0.4f" % TNR) + print("False positive rate: %0.4f" % FPR) + print('support: ' , support) + return individual_metrics + +def collecting_individual_metrics_singletask(model, x, parameters, rhythm, out_message=False): + """ + """ + individual_metrics = {} + for i in np.unique(parameters['ID']): + # Sub-selecting individuals + p_indx = np.where(parameters['ID'] == i)[0] + x_pID = x[p_indx] + parameters_pID = parameters.iloc[p_indx] + rhythm_pID = rhythm[p_indx] + # Rhythmn predictions + predictions_r = model.predict(x_pID) + # If the estimated prob of AF(column 1) is higher than the estimated prob of non-AF(column 0) + # than consider the window as an AF window. + y_predictions = np.argmax(predictions_r, axis=1) + y_truth = np.argmax(rhythm_pID, axis=1) + # Confusion_matrix + cf = confusion_matrix(y_truth, y_predictions, labels=[0, 1]) + TN, FP, FN, TP = cf.ravel() + support = TN+FP+FN+TP + # Sensitivity, recall, or true positive rate + TPR = TP/(TP+FN) + # Specificity or true negative rate + TNR = TN/(TN+FP) + # Fall out or false positive rate + FPR = FP/(FP+TN) + # False negative rate + FNR = FN/(TP+FN) + # F1 score + f1 = metrics.f1_score(y_truth, y_predictions, average=None) + # selecting f1score for positive case if present + f1_pos = f1[1] if len(f1) > 1 else f1[0] + # Area under precision recall curve + #auprc = metrics.average_precision_score(y_test_truth, rhythm_pID[:,1]) + if (TP + FN) > 0: + individual_metrics[i] = [TPR, np.NaN, np.NaN, FNR, f1_pos, support] + if (FP + TN) > 0: + individual_metrics[i] = [np.NaN, TNR, FPR, np.NaN, np.NaN, support] + if out_message: + print("PATIENT :", i , '\n') + print(pd.DataFrame(cf).rename(columns={0: "Predicted Non-AF", 1:" Predicted AF"}, index={0: "True Non-AF", 1:"True AF"})) + if (TP + FN) > 0: + print("Sensitivity/Recall: %0.4f" % TPR) + print("False negative rate: %0.4f" % FNR) + print("F1 score: %0.4f" % f1_pos) + print('support: ' , support) + if (FP + TN) > 0: + print("Specificity: %0.4f" % TNR) + print("False positive rate: %0.4f" % FPR) + print('support: ' , support) + return individual_metrics