deepthiaj/Electro_oneAPI

1import pandas as pd2import numpy as np3import wfdb4import ast5 6def load_raw_data(df, sampling_rate, path):7    if sampling_rate == 100:8        data = [wfdb.rdsamp(path+f) for f in df.filename_lr]9    else:10        data = [wfdb.rdsamp(path+f) for f in df.filename_hr]11    data = np.array([signal for signal, meta in data])12    return data13 14path = 'path/to/ptbxl/'15sampling_rate=10016 17# load and convert annotation data18Y = pd.read_csv(path+'ptbxl_database.csv', index_col='ecg_id')19Y.scp_codes = Y.scp_codes.apply(lambda x: ast.literal_eval(x))20 21# Load raw signal data22X = load_raw_data(Y, sampling_rate, path)23 24# Load scp_statements.csv for diagnostic aggregation25agg_df = pd.read_csv(path+'scp_statements.csv', index_col=0)26agg_df = agg_df[agg_df.diagnostic == 1]27 28def aggregate_diagnostic(y_dic):29    tmp = []30    for key in y_dic.keys():31        if key in agg_df.index:32            tmp.append(agg_df.loc[key].diagnostic_class)33    return list(set(tmp))34 35# Apply diagnostic superclass36Y['diagnostic_superclass'] = Y.scp_codes.apply(aggregate_diagnostic)37 38# Split data into train and test39test_fold = 1040# Train41X_train = X[np.where(Y.strat_fold != test_fold)]42y_train = Y[(Y.strat_fold != test_fold)].diagnostic_superclass43# Test44X_test = X[np.where(Y.strat_fold == test_fold)]45y_test = Y[Y.strat_fold == test_fold].diagnostic_superclass46