52 lines
2.1 KiB
Python
52 lines
2.1 KiB
Python
import pandas as pd
|
||
import numpy as np
|
||
from pathlib import Path
|
||
|
||
def find_TPeak(data,peaks,th=50):
|
||
"""
|
||
找出真实的J峰或R峰
|
||
:param data: BCG或ECG数据
|
||
:param peaks: 初步峰值（从label中导出的location_R）
|
||
:param th: 范围阈值
|
||
:return: 真实峰值
|
||
"""
|
||
return_peak = []
|
||
for peak in peaks:
|
||
if peak>len(data):continue
|
||
min_win,max_win = max(0,int(peak-th)),min(len(data),int(peak+th))
|
||
return_peak.append(np.argmax(data[min_win:max_win])+min_win)
|
||
return np.array(return_peak)
|
||
|
||
def Rpeak_Detection(input_dir, fs, th1):
|
||
raw_ecg = pd.read_csv(Path(input_dir) / "filter_ecg.txt").to_numpy().reshape(-1)
|
||
# raw_ecg = raw_ecg[200*sample_rate:]
|
||
#######################限制幅值处理############################################
|
||
# for i in range(len(raw_ecg)):
|
||
# if raw_ecg[i] > 300 or raw_ecg[i] < -300:
|
||
# raw_ecg[i] = 0
|
||
##############################################################################
|
||
all_R_peak = np.array([])
|
||
|
||
##############################切割处理##########################################
|
||
for file_num in range(1, len(list((Path(input_dir) / "label").rglob("*location_J.txt"))) + 1):
|
||
R_peak_dir = Path(input_dir) / "label" / (str(file_num) + 'location_J.txt')
|
||
R_peak = np.array(pd.read_csv(R_peak_dir, header=None)).reshape(-1)
|
||
R_peak = R_peak + fs * 3600 * (file_num - 1)
|
||
all_R_peak = np.append(all_R_peak, R_peak)
|
||
##############################################################################
|
||
all_R_peak = find_TPeak(raw_ecg, all_R_peak, th=int(th1 * fs / 1000))
|
||
|
||
RR_Interval = np.full(len(all_R_peak) - 1, np.nan)
|
||
|
||
for i in range(len(all_R_peak) - 1):
|
||
RR_Interval[i] = all_R_peak[i + 1] - all_R_peak[i]
|
||
|
||
RRIV = np.full(len(RR_Interval) - 1, np.nan)
|
||
for i in range(len(RR_Interval) - 1):
|
||
RRIV[i] = RR_Interval[i + 1] - RR_Interval[i]
|
||
|
||
Interval = np.full(len(raw_ecg), np.nan)
|
||
for i in range(len(all_R_peak) - 1):
|
||
Interval[all_R_peak[i]: all_R_peak[i + 1]] = all_R_peak[i + 1] - all_R_peak[i]
|
||
|
||
return all_R_peak.reshape(-1) |