37 lines
1.4 KiB
Python
37 lines
1.4 KiB
Python
import utils
|
||
import pandas as pd
|
||
import numpy as np
|
||
from scipy import signal
|
||
|
||
def normalize_resp_signal(resp_signal: np.ndarray, resp_fs, movement_mask, enable_list):
|
||
# 根据呼吸信号的幅值改变区间，对每段进行Z-Score标准化
|
||
normalized_resp_signal = np.zeros_like(resp_signal)
|
||
# 全部填成nan
|
||
normalized_resp_signal[:] = np.nan
|
||
|
||
resp_signal_no_movement = resp_signal.copy()
|
||
|
||
|
||
resp_signal_no_movement[np.array(movement_mask == 1).repeat(resp_fs)] = np.nan
|
||
|
||
|
||
for i in range(len(enable_list)):
|
||
enable_start = enable_list[i][0] * resp_fs
|
||
enable_end = enable_list[i][1] * resp_fs
|
||
segment = resp_signal_no_movement[enable_start:enable_end]
|
||
|
||
# print(f"Normalizing segment {i+1}/{len(enable_list)}: start={enable_start}, end={enable_end}, length={len(segment)}")
|
||
|
||
segment_mean = np.nanmean(segment)
|
||
segment_std = np.nanstd(segment)
|
||
if segment_std == 0:
|
||
raise ValueError(f"segment_std is zero! segment_start: {enable_start}, segment_end: {enable_end}")
|
||
|
||
# 同下一个enable区间的体动一起进行标准化
|
||
if i <= len(enable_list) - 2:
|
||
enable_end = enable_list[i + 1][0] * resp_fs
|
||
raw_segment = resp_signal[enable_start:enable_end]
|
||
normalized_resp_signal[enable_start:enable_end] = (raw_segment - segment_mean) / segment_std
|
||
|
||
return normalized_resp_signal
|