HeartGPT / Heart_PT_generate.py 공부하기

https://github.com/harryjdavies/HeartGPT/blob/main/Heart_PT_generate.py

HeartGPT/Heart_PT_generate.py at main · harryjdavies/HeartGPT

 

위 코드 해석해보자.

 

이 코드는 심장 생체 신호(ECG 및 PPG)를 처리하는 Transformer 기반 모델(Heart_GPT_Model)을 정의하고, 이를 사용하여 생체 신호 데이터를 기반으로 새로운 신호 데이터를 생성하는 코드이다.

 

전체적인 흐름을 설명하면 다음과 같다.

 

 

---

1. 라이브러리 임포트

import torch
import torch.nn as nn
from torch.nn import functional as F
import scipy.io
import numpy as np
import pandas as pd

 

 

• torch, torch.nn, F: PyTorch 기반의 신경망을 구축하고 학습하는 데 필요한 라이브러리다.
• scipy.io: 데이터 처리를 위해 사용되지만, 코드에서는 직접 사용되지 않음.
• numpy, pandas: 데이터를 다루는 데 필요한 라이브러리.

 

---

 

2. 모델 설정

model_config = 'ECG_PT' # 'ECG_PT' 또는 'PPG_PT' 선택
block_size = 500  # 컨텍스트 길이
n_embd = 64  # 임베딩 차원
n_head = 8  # 멀티 헤드 어텐션의 헤드 개수
n_layer = 8  # 트랜스포머 블록 개수
dropout = 0.2  # 드롭아웃 비율

 

 

 

• 모델이 ECG(심전도) 또는 PPG(광용적맥파) 데이터를 처리할 수 있도록 구성됩니다.
• block_size = 500 → 모델이 한 번에 처리할 수 있는 입력 길이는 500개 샘플.
• n_embd, n_head, n_layer, dropout → 트랜스포머 모델의 주요 하이퍼파라미터.

 

모델 경로 설정

model_path_ppg = "D:/HeartGPTModels/PPGPT_500k_iters.pth"
model_path_ecg = "D:/HeartGPTModels/ECGPT_560k_iters.pth"
device = 'cuda' if torch.cuda.is_available() else 'cpu'

 

 

 

• 모델 가중치 파일의 경로 지정 (PPGPT_500k_iters.pth & ECGPT_560k_iters.pth).
• CUDA(그래픽카드) 사용 가능 여부 확인.

 

모델 종류에 따른 설정

if model_config == 'PPG_PT':
    vocab_size = 102  # PPG 모델의 어휘 크기
    model_path = model_path_ppg
    context_path = 'D:/HeartGPTModels/example_context_PPG.csv'
elif model_config == 'ECG_PT':
    vocab_size = 101  # ECG 모델의 어휘 크기
    model_path = model_path_ecg
    context_path = 'D:/HeartGPTModels/example_context_ECG.csv'

 

 

• PPG_PT 모델을 사용할 경우: vocab_size = 102, PPGPT_500k_iters.pth 사용.
• ECG_PT 모델을 사용할 경우: vocab_size = 101, ECGPT_560k_iters.pth 사용.

---

 

3. 생체 신호 데이터 토큰화 함수

def tokenize_biosignal(data):
    shape = data.shape
    if len(shape) > 1 and shape[0] > shape[1]:
        data = data.T  # 세로형 데이터를 가로형으로 변환

    if data.shape[1] > 500:
        data = data[:, -500:]  # 마지막 500개의 데이터만 선택

    data_min = np.min(data)
    data_max = np.max(data)
    data_scaled = (data - data_min) / (data_max - data_min)  # 0~1로 정규화

    data_scaled *= 100  # 0~100 범위로 조정
    data_rounded = np.round(data_scaled)  # 정수화

    return data_rounded

 

• 데이터를 0~100 사이의 정수 값으로 변환하여 모델 입력으로 사용할 수 있도록 전처리하는 함수.

---

4. 트랜스포머 기반 모델 정의 (Heart_GPT_Model)

• Transformer 블록 구조
  • Head: 단일 Self-Attention 헤드
  • MultiHeadAttention: 여러 개의 Head를 병렬 실행하여 정보를 조합
  • FeedForward: 어텐션 후 데이터를 변형하는 MLP 레이어
  • Block: 트랜스포머 블록 (레이어 정규화 + 어텐션 + 피드포워드)
• 모델 구조

class Heart_GPT_Model(nn.Module):

    def __init__(self):
        super().__init__()
        self.token_embedding_table = nn.Embedding(vocab_size,n_embd)
        self.position_embedding_table = nn.Embedding(block_size, n_embd)
        self.blocks = nn.Sequential(*[Block(n_embd, n_head) for _ in range(n_layer)])
        self.ln_f = nn.LayerNorm(n_embd)
        self.lm_head = nn.Linear(n_embd, vocab_size)

    def forward(self, idx, targets=None):
        B, T = idx.shape
        tok_emb = self.token_embedding_table(idx)
        pos_emb = self.position_embedding_table(torch.arange(T, device=device))
        x = tok_emb + pos_emb
        x = self.blocks(x)
        x = self.ln_f(x)
        logits = self.lm_head(x)

        if targets is None:
            loss = None
        else:
            B, T, C = logits.shape
            logits = logits.view(B*T, C)
            targets = targets.view(B*T)
            loss = F.cross_entropy(logits, targets)

        return logits, loss

 

 

• 토큰 임베딩 및 위치 임베딩을 사용하여 Transformer 모델을 구축.
• Cross-Entropy 손실을 계산하여 학습할 수 있도록 설계.

---

 

5. 모델 로드

model = Heart_GPT_Model()
model.load_state_dict(torch.load(model_path))
model.eval()
m = model.to(device)

 

 

• 저장된 모델 가중치를 불러오고 eval() 모드로 설정.

---

 

6. 데이터 로드 및 변환

df = pd.read_csv(context_path, header=None)
data = df.values
data_tokenised = tokenize_biosignal(data)
example_context_tensor = torch.tensor(data_tokenised, dtype=torch.long, device = device)

 

• CSV에서 ECG 또는 PPG 데이터를 불러와 토큰화 후 PyTorch 텐서로 변환.

---

 

7. 모델을 이용한 시퀀스 생성

print('Generating new tokens')
output = (m.generate(example_context_tensor, max_new_tokens=500)[0].tolist())

 

• generate() 함수를 사용하여 새로운 500개의 토큰을 생성.

---

8. 결과 저장

output_df = pd.DataFrame(output)
output_df.to_csv('D:/HeartGPTModels/model_output.csv', index=False, header=False)

data_tokenised = np.transpose(data_tokenised).tolist()
data_tokenised_df = pd.DataFrame(data_tokenised)
data_tokenised_df = data_tokenised_df.astype(int)
data_tokenised_df.to_csv('D:/HeartGPTModels/tokenised_context.csv', index=False, header=False)

print('Generation saved to CSV')

 

• 모델이 생성한 데이터를 CSV 파일로 저장.

---

 

정리

1. ECG 또는 PPG 신호 데이터를 불러와 토큰화한다.

2. Transformer 모델(Heart_GPT_Model)로 학습된 가중치를 로드한다.

3. 모델을 이용해 새로운 500개의 토큰을 생성한다.

4. 생성된 데이터와 토큰화된 입력 데이터를 CSV 파일로 저장한다.