SleepMaMi / data /data_utils.py
KeondoPark
Initial commit for public release
35f60b9
Raw History Blame Contribute Delete
2.64 kB
import os
import numpy as np
from scipy.signal import butter, sosfilt, lfilter, iirnotch
from scipy.signal import find_peaks
def butter_bandpass(lowcut, highcut, fs, order=8):
low = lowcut
high = highcut
sos = butter(order, [low, high], btype='band', fs=fs, output='sos', analog=False)
return sos
def butter_highpass(lowcut, fs, order=8):
low = lowcut
sos = butter(order, low, btype='highpass', fs=fs, output='sos', analog=False)
return sos
def butter_lowpass(highcut, fs, order=8):
high = highcut
sos = butter(order, high, btype='lowpass', fs=fs, output='sos', analog=False)
return sos
def butter_bandpass_filter(data, lowcut, highcut, fs, order=8):
if highcut > fs/2:
sos = butter_highpass(lowcut, fs, order=order)
elif lowcut <= 0:
sos = butter_lowpass(highcut, fs, order=order)
else:
sos = butter_bandpass(lowcut, highcut, fs, order=order)
#y = lfilter(b, a, data)
y = sosfilt(sos, data)
return y
def correct_ecg_inversion(ecg_signal, sampling_rate):
"""
ECG ์‹ ํ˜ธ๊ฐ€ ๋’ค์ง‘ํ˜”๋Š”์ง€ ๊ฐ์ง€ํ•˜๊ณ  ์˜ฌ๋ฐ”๋ฅด๊ฒŒ ์ˆ˜์ •ํ•ฉ๋‹ˆ๋‹ค.
Args:
ecg_signal (np.array): ECG ์‹ ํ˜ธ ๋ฐ์ดํ„ฐ.
sampling_rate (int): ์ƒ˜ํ”Œ๋ง ์ฃผํŒŒ์ˆ˜ (Hz).
Returns:
np.array: ์ˆ˜์ •๋œ ECG ์‹ ํ˜ธ.
bool: ๋’ค์ง‘ํž˜์ด ๊ฐ์ง€๋˜์—ˆ๋Š”์ง€ ์—ฌ๋ถ€.
"""
# QRS๊ตฐ์˜ RํŒŒ๋ฅผ ์ฐพ๊ธฐ ์œ„ํ•ด ์‹ ํ˜ธ์˜ ์–‘์ˆ˜ ํ”ผํฌ๋ฅผ ์ฐพ์Œ
# prominence: ํ”ผํฌ๊ฐ€ ์ฃผ๋ณ€๋ณด๋‹ค ์–ผ๋งˆ๋‚˜ ํŠ€์–ด๋‚˜์™”๋Š”์ง€, distance: ํ”ผํฌ ๊ฐ„ ์ตœ์†Œ ๊ฑฐ๋ฆฌ
positive_peaks, _ = find_peaks(ecg_signal, prominence=0.1, distance=0.3 * sampling_rate)
# QRS๊ตฐ์˜ SํŒŒ(๋’ค์ง‘ํžŒ RํŒŒ)๋ฅผ ์ฐพ๊ธฐ ์œ„ํ•ด ์‹ ํ˜ธ๋ฅผ ๋’ค์ง‘์–ด์„œ ํ”ผํฌ๋ฅผ ์ฐพ์Œ
negative_peaks, _ = find_peaks(-ecg_signal, prominence=0.1, distance=0.3 * sampling_rate)
# ํ”ผํฌ๊ฐ€ ํ•˜๋‚˜๋„ ๊ฐ์ง€๋˜์ง€ ์•Š์œผ๋ฉด ์›๋ณธ ์‹ ํ˜ธ ๋ฐ˜ํ™˜
if len(positive_peaks) == 0 and len(negative_peaks) == 0:
return ecg_signal, False
# ์–‘์ˆ˜ ํ”ผํฌ์™€ ์Œ์ˆ˜ ํ”ผํฌ์˜ ํ‰๊ท  ์ง„ํญ ๊ณ„์‚ฐ
mean_pos_amp = np.mean(ecg_signal[positive_peaks]) if len(positive_peaks) > 0 else 0
mean_neg_amp = np.mean(ecg_signal[negative_peaks]) if len(negative_peaks) > 0 else 0
# ๋งŒ์•ฝ ์Œ์ˆ˜ ํ”ผํฌ์˜ ์ ˆ๋Œ€๊ฐ’์ด ์–‘์ˆ˜ ํ”ผํฌ๋ณด๋‹ค ํฌ๋ฉด ๋’ค์ง‘ํžŒ ๊ฒƒ์œผ๋กœ ํŒ๋‹จ
is_inverted = abs(mean_neg_amp) > abs(mean_pos_amp)
if is_inverted:
# print("๐Ÿ’ก ECG ์‹ ํ˜ธ๊ฐ€ ๋’ค์ง‘ํžŒ ๊ฒƒ์„ ๊ฐ์ง€ํ•˜์—ฌ ์ˆ˜์ •ํ•ฉ๋‹ˆ๋‹ค.")
return ecg_signal * -1, True
else:
# print("โœ… ECG ์‹ ํ˜ธ๊ฐ€ ์ •์ƒ ๋ฐฉํ–ฅ์ž…๋‹ˆ๋‹ค.")
return ecg_signal, False