데이터사이언스/인공지능

PTB-XL ECG Dataset Description

누군가의 이야기 2022. 5. 14. 14:38
728x90

데이터셋 소개

https://physionet.org/content/ptb-xl/1.0.1/

 

PTB-XL, a large publicly available electrocardiography dataset v1.0.1

 

physionet.org


- 공공 이용 가능한 데이터

 

- 1989 년부터 1996 년까지 Schiller AG라는 ECG data 기록기를 통해 얻은 데이터

 

- 18885명의 환자로부터 21837 개 데이터 수집

 

- 10초간 측정 기록

 

-  12-lead ECG-waveform dataset

 

- 2명의 심장병 전문의가 레이블링함

 

- 500Hz (5000) / 100Hz (1000) 2가지 데이터 포함되어 있기 때문에 선택적으로 사용 가능

 


위 데이터를 가지고 심근경색 감지하는 인공지는 기반 모델을 구축해보는 프로젝트를 수행했었다.

 

데이터베이스 csv파일을 열어보면 환자를 식별할 수 있는 정보는 제거되었고,

 

수많은 column이 존재하는 걸 볼 수 있는데

 

이중에 scp_codes(L column)가 핵심이다.

 

해당 데이터로 진단된 심장질환이 레이블링 되어있는데, 해당 질환의 확률이 딕셔너리 형태로 함께 제공된다.

 

레이블 심장 질환은 5개의 주요 클래스와 24개의 하위 클래스로 구성되어 있는데, 아래 도표와 같다.

 

https://www.nature.com/articles/s41597-020-0495-6

여기서 나는 MI(Myocardial Infarction) 심근경색 클래스에 해당하는 하위 클래스를 Target Label으로 설정하여 프로젝트를 진행하였다.

 


 

데이터 로드와 전처리를 용이하게 해주는 예제 파이썬 파일이 함께 들어있어 매우 편리하다.

 

본 데이터는 이미 10개의 fold로 나뉘어 있는데, 아래 코드를 바탕으로 train / test set 나누는 것도 용이하다.

 

import pandas as pd
import numpy as np
import wfdb
import ast

def load_raw_data(df, sampling_rate, path):
    if sampling_rate == 100:
        data = [wfdb.rdsamp(path+f) for f in df.filename_lr]
    else:
        data = [wfdb.rdsamp(path+f) for f in df.filename_hr]
    data = np.array([signal for signal, meta in data])
    return data

path = 'path/to/ptbxl/'
sampling_rate=100

# load and convert annotation data
Y = pd.read_csv(path+'ptbxl_database.csv', index_col='ecg_id')
Y.scp_codes = Y.scp_codes.apply(lambda x: ast.literal_eval(x))

# Load raw signal data
X = load_raw_data(Y, sampling_rate, path)

# Load scp_statements.csv for diagnostic aggregation
agg_df = pd.read_csv(path+'scp_statements.csv', index_col=0)
agg_df = agg_df[agg_df.diagnostic == 1]

def aggregate_diagnostic(y_dic):
    tmp = []
    for key in y_dic.keys():
        if key in agg_df.index:
            tmp.append(agg_df.loc[key].diagnostic_class)
    return list(set(tmp))

# Apply diagnostic superclass
Y['diagnostic_superclass'] = Y.scp_codes.apply(aggregate_diagnostic)

# Split data into train and test
test_fold = 10
# Train
X_train = X[np.where(Y.strat_fold != test_fold)]
y_train = Y[(Y.strat_fold != test_fold)].diagnostic_superclass
# Test
X_test = X[np.where(Y.strat_fold == test_fold)]
y_test = Y[Y.strat_fold == test_fold].diagnostic_superclass
728x90