import binascii import numpy as np import pandas as pd import scipy.stats as st import os import heartpy as hp import directory import time import re import PLOT import hrvanalysis import json from directory import N_DATA_PPG import sys import json import ast class PPG: def __init__(self, file, RESULT_DIR, age='20', device='soso', depressed=0, unrest=0, people_file=None, reportLanguage="kr"): self.file = file self.RESULT_DIR = RESULT_DIR self.people_file = people_file self.device = device self.samplefreq = directory.cal_samplefreq(self.file, self.device) # soso 24 구분을 위해 file 인수 추가 (아직 live 추가 x) self.real_age = age # 웹에서 던져주는 실제 나이 (설문지 구분을 위해 필요) self.age = directory.age_modify(age) # z-score 계산을 위한 연령대 그룹 self.raw = None self.filtered = None self.peaks = None self.sdnn = None self.fd = None self.zscore_ans = None self.ab = None self.sns = None self.si = None self.zscore_si = None self.sr = None self.zscore_sr = None self.depressed = depressed self.unrest = unrest self.depressed_score = None self.unrest_score = None self.ppg_score = None self.ppg_emoji = None self.data_quality_ppg = "high" self.ppg_ebsi = None self.reportLanguage = reportLanguage if self.reportLanguage == "en": PLOT.BASE_DIR = directory.BASE_DIR_EN self.report = {'ppg_code': -2006, 'ppgScore': 0, 'ppgOnlyScore': 0, 'dataQuality_ppg': 'bad', 'ppg_subtype': None} # 레포트 출력을 위한 결과값 self.viewer_data = {'ppg': []} self.dq = {'dataQuality_ppg': 'bad'} try: self.msn = sys.argv[17]#검사기관 조건처리 위해 추가 25.01.14 print(sys.argv[17]) except: self.msn = sys.argv[3] print(sys.argv[3])#onestop dataquality 검사기관 조건처리 위해 추가 25.01.31 def load_file(self): _, extension = os.path.splitext(self.file) if (self.device == "soso") and (extension == ".arc"): # soso device format f = open(self.file, "rb") t = f.read() tt = binascii.hexlify(t) by = re.findall('a0\w{62}c0', str(tt)) temp = [] for i in by: ppg = i[16:22] ppg_trans = directory.transHex2Deci(ppg[0:2], ppg[2:4], ppg[4:6]) temp.append(ppg_trans) f.close() data = [temp[i] for i in range(0, len(temp))] if ((self.device == "soso") or (self.device == "laxtha")) and (extension == ".txt"): f = open(self.file, "r", encoding='euc-kr') temp = [] while True: line = f.readline() if not line: break temp.append(line) f.close() data = [temp[i].split()[2] for i in range(7, len(temp))] # header 제외 if ((self.device == "arduino") and (extension == ".txt")): f = open(self.file, "r", encoding='euc-kr') temp = [] while True: line = f.readline() if not line: break temp.append(line) f.close() data = [temp[i].split('\n')[0] for i in range(1, len(temp))] if (self.device == "cgx") and (extension == ".txt"): data_df = pd.read_table(self.file, skiprows=[0, 1, 2, 3], sep="\t", low_memory=False) data_df = data_df[data_df.columns[48]] data = data_df.values if (self.device == "mini") and (extension == ".txt"): f = open(self.file, "r", encoding='euc-kr') temp = [] while True: line = f.readline() if not line: break temp.append(line) f.close() data = [temp[i].split()[1] for i in range(0, len(temp))] if (self.device == "light_soso") and (extension == ".txt"): # soso device format if self.file.split('/')[-1].split('_')[1] =='02': # soso24 파일 split 했을 때 02면 24로 타고 아니면 기존 19/23버전 탐 (live 추가 x ) print('soso_24 version') f = open(self.file, "rb") t = f.read() # tt = binascii.hexlify(t) by = re.findall('a0\w{62}c0', str(t)) temp = [] for i in by: ppg = i[16:22] ppg_trans = directory.transHex2Deci_24(ppg[0:2], ppg[2:4], ppg[4:6]) temp.append(ppg_trans) f.close() # soso24: 약 63초마다 깨진 패킷 1개(4ms)가 수백만 값으로 읽혀 필터를 오염시킴 -> 주변 값으로 보간 temp = self.remove_spike(temp) else: print("soso 19/23") # 확인하려고 print 찍음 (live 동기화 하지 말것 ) f = open(self.file, "rb") t = f.read() # tt = binascii.hexlify(t) by = re.findall('a0\w{62}c0', str(t)) temp = [] for i in by: ppg = i[16:22] ppg_trans = directory.transHex2Deci(ppg[0:2], ppg[2:4], ppg[4:6]) temp.append(ppg_trans) f.close() data = [temp[i] for i in range(0, len(temp))] if self.device == 'linkband': self.samplefreq = 50.0 with open(self.file, 'r', encoding='utf-8') as f: js = json.load(f) ir = js.get("ppgIr", []) red = js.get("ppgRed", []) leadoff_raw = js.get('eegLeadOff', []) #리드오프 오류처리 26.01.28 if isinstance(leadoff_raw, str): leadoff = json.loads(leadoff_raw) else: leadoff = leadoff_raw total = len(leadoff) num_off = sum(1 for x in leadoff if x == 1) off_ratio = num_off / total if total > 0 else 0 #리드오프 50퍼센트 초과일 경우 오류 발생 if off_ratio > 0.5: raise Exception("Lead-Off Error") if isinstance(ir, str): ir = ast.literal_eval(ir) if isinstance(red, str): red = ast.literal_eval(red) # 이마 측정: IR 우선, 비어있으면 Red 폴백 data = ir if (ir and len(ir) > 0) else red data = pd.DataFrame(data, columns=['PPG']) data = data.dropna() data = data.astype('float') # * 1e-6 # if (self.device == "light_soso") and (extension == ".txt"): # light_soso의 경우 기준을 엄격히 했더니 품질이 다 안좋게 나와서 주석처리 24.05.24 # data = data.loc[(abs(data['PPG']) < 500)] # self.raw = data['PPG'] # else: # self.raw = data['PPG'] self.raw = data['PPG'] return self.raw.values @staticmethod def remove_spike(values, window=11, threshold=500): # 국소 중앙값에서 threshold(단위: 변환값, 약 20000 count) 이상 벗어난 단일 샘플을 선형 보간 s = pd.Series(values, dtype='float') med = s.rolling(window, center=True, min_periods=1).median() bad = (s - med).abs() > threshold if bad.any(): print("soso24 spike removed:", int(bad.sum())) s[bad] = np.nan s = s.interpolate(limit_direction='both') return s.tolist() def viewr_preprocess(self): raw = self.raw.values ppg_x = np.arange(0, len(raw)/self.samplefreq, 1/self.samplefreq) ppg_df = pd.DataFrame(np.array([ppg_x, raw]).T, columns=['time', 'ppg']) return ppg_df def preprocess(self, tmax=305): print("preprocess") # self.raw = self.notch_pass_filter(data=data, center=60, interval=4, normalized=True) if self.device == "soso": self.filtered = hp.filter_signal(self.raw, [0.5, 5], self.samplefreq, order=2, filtertype='bandpass',return_top=True) else: self.filtered = hp.filter_signal(self.raw, [0.5, 2], self.samplefreq, order=2, filtertype='bandpass', return_top=True) #if self.filtered.shape[0] < self.samplefreq * 180: # 3분 미만이면 -> 1,3분 도입으로 시간기준 필요 x 주석처리 25.01.13 #self.data_quality_ppg = "low" #raise Exception('ppg_data_quality bad') #elif self.filtered.shape[0] < self.samplefreq * 300: # 5분 미만이면 #self.data_quality_ppg = "middle" self.filtered = self.filtered[:int(self.samplefreq*tmax)] def cal_time_freq_domain(self): wd, m = hp.process(self.filtered, sample_rate=self.samplefreq, windowsize=1.5, calc_freq=False, hampel_correct=False, high_precision=True, reject_segmentwise=False, clean_rr=True, clean_rr_method='z-score', interp_clipping=True) # bad rri ratio ratio = len(np.where(np.array(wd['RR_masklist']) == 1)[0]) / len(wd['RR_list']) self.sdnn = m['sdnn'] self.hrt = m['bpm'] self.rri = wd['RR_list_cor'] self.report['bpm'] = round(self.hrt,2) self.rmssd = m['rmssd'] if (self.sdnn > 135) or (ratio > 0.4) or (self.hrt > 130): self.data_quality_ppg = "low" raise Exception('ppg_data_quality bad') if (ratio < 0.2): #기존 0.1에서 0.2로 완화 25.05.30 self.data_quality_ppg = "high" elif (ratio < 0.3):#기존 0.2에서 0.3으로 완화 25.05.30 self.data_quality_ppg = "middle" else: self.data_quality_ppg = "low" self.report['dataQuality_ppg'] = self.data_quality_ppg self.dq['dataQuality_ppg'] = self.data_quality_ppg print('DROP Ratio:', ratio, "SDNN:", self.sdnn) # freq domain fd = hrvanalysis.get_frequency_domain_features(self.rri, method="welch") tp = fd['total_power'] vlf = fd['vlf'] lf = fd['lf'] hf = fd['hf'] lf_nu = fd['lfnu'] hf_nu = fd['hfnu'] rlfhf = lf_nu / hf_nu tp = np.log(tp) vlf = np.log(vlf) lf = np.log(lf) hf = np.log(hf) self.fd = pd.DataFrame([[tp, vlf, lf, hf, lf_nu, hf_nu, rlfhf]], columns=[['TP', 'VLF', 'LF', 'HF', 'nLF', 'nHF', 'RLFHF']]) ''' 자율 신경 활성도 계산 ''' def cal_ans_activity(self): tp = self.fd['TP'].values[0][0] vlf = self.fd['VLF'].values[0][0] lf = self.fd['LF'].values[0][0] hf = self.fd['HF'].values[0][0] zscore_tp = (tp - float(N_DATA_PPG[''.join(['TP_'+self.age+'_avg'])])) / float(N_DATA_PPG[''.join(['TP_', self.age, '_std'])]) zscore_vlf = (vlf - float(N_DATA_PPG[''.join(['VLF_', self.age, '_avg'])])) / float(N_DATA_PPG[''.join(['VLF_', self.age, '_std'])]) zscore_lf = (lf - float(N_DATA_PPG[''.join(['LF_', self.age, '_avg'])])) / float(N_DATA_PPG[''.join(['LF_', self.age, '_std'])]) zscore_hf = (hf - float(N_DATA_PPG[''.join(['HF_', self.age, '_avg'])])) / float(N_DATA_PPG[''.join(['HF_', self.age, '_std'])]) zscore_ans = [zscore_tp, zscore_vlf, zscore_lf, zscore_hf] ratio_ans = list(st.norm.cdf(zscore_ans) * 100) self.zscore_ans = zscore_ans self.report['ans'] = [int(v) for v in ratio_ans] # plot PLOT.plot_ans_activity(ratio_ans, self.RESULT_DIR) ''' 자율 신경 균형도 계산 ''' def cal_ans_balance(self): sns = self.fd['nLF'].values[0][0] pns = self.fd['nHF'].values[0][0] self.ab = [sns, pns] self.sns = self.ab[0] self.report['ab'] = (np.round(self.ab,1)).tolist() # plot PLOT.plot_ans_balance(self.ab, self.RESULT_DIR) ''' 스트레스 지수 계산 ''' def cal_stress_index(self): self.si = self.sns / self.sdnn self.zscore_si = (np.sqrt(self.si) - float(N_DATA_PPG[''.join(['SI_', self.age, '_avg'])])) / float(N_DATA_PPG[''.join(['SI_', self.age, '_std'])]) ratio_si = st.norm.cdf(self.zscore_si) * 100 self.report['si']=round(ratio_si) # plot PLOT.plot_stress_index(ratio_si, self.RESULT_DIR, reportLanguage=self.reportLanguage) ''' 스트레스 저항도 계산 ''' def cal_stress_resistance(self): self.sr = self.sdnn self.zscore_sr = (self.sr - float(N_DATA_PPG[''.join(['SR_', self.age, '_avg'])])) / float(N_DATA_PPG[''.join(['SR_', self.age, '_std'])]) ratio_sr = st.norm.cdf(self.zscore_sr) * 100 self.report['sr']=round(ratio_sr) # plot PLOT.plot_stress_resistance(ratio_sr, self.RESULT_DIR, reportLanguage=self.reportLanguage) ''' 설문 우울점수 ''' def cal_depress(self, reportLanguage="kr"): x_val = 0 depressed_score = 0 if int(self.real_age) < 14: # 설문지 구별을 위해 실제 나이 사용 depressed_score = 0 x_val = 0 if self.depressed == 0: depressed_score = 0 x_val = 0.5 elif self.depressed == 1: depressed_score = 3 x_val = 1.5 elif self.depressed == 2: depressed_score = 6 x_val = 2.4 elif self.depressed == 3: depressed_score = 10 x_val = 3.4 elif self.depressed == 4: depressed_score = 15 x_val = 4.45 else: depressed_score = 15 x_val = 4.45 else: if self.reportLanguage == "en": if self.depressed < 5: x_val = self.depressed / 5 * 0.95 depressed_score = 0 elif (5 <= self.depressed) and (self.depressed < 10): x_val = (self.depressed - 5) / 5 * 0.95 + 0.95 depressed_score = 3 elif (10 <= self.depressed) and (self.depressed < 15): x_val = (self.depressed - 10) / 4 * 0.95 + 1.9 depressed_score = 6 elif (15 <= self.depressed) and (self.depressed < 20): x_val = (self.depressed - 15) / 5 * 0.95 + 2.95 depressed_score = 10 elif (20 <= self.depressed) and (self.depressed < 28): x_val = (self.depressed - 20) / 8 * 1 + 3.9 depressed_score = 15 else: x_val = (self.depressed - 20) / 8 * 1 + 3.9 depressed_score = 15 else: if self.depressed < 9: x_val = self.depressed / 9 * 0.95 depressed_score = 0 elif (9 <= self.depressed) and (self.depressed < 13): x_val = (self.depressed - 9) / 4 * 0.95 + 0.95 depressed_score = 3 elif (13 <= self.depressed) and (self.depressed < 17): x_val = (self.depressed - 13) / 4 * 0.95 + 1.9 depressed_score = 6 elif (17 <= self.depressed) and (self.depressed < 21): x_val = (self.depressed - 17) / 4 * 0.95 + 2.95 depressed_score = 10 elif (21 <= self.depressed) and (self.depressed < 60): x_val = (self.depressed - 21) / 39 * 1 + 3.9 depressed_score = 15 else: x_val = (self.depressed - 21) / 39 * 1 + 3.9 depressed_score = 15 self.report['dep'] = round(self.depressed,2) return x_val, depressed_score ''' 설문 불안점수 ''' def cal_unrest(self, reportLanguage="kr"): x_val = 0 unrest_score = 0 if int(self.real_age) < 14: # 설문지 구별을 위해 실제 나이 사용 unrest_score = 0 if self.unrest == 0: unrest_score = 0 x_val = 0.5 elif self.unrest == 1: unrest_score = 3 x_val = 1.5 elif self.unrest == 2: unrest_score = 6 x_val = 2.4 elif self.unrest == 3: unrest_score = 10 x_val = 3.4 elif self.unrest == 4: unrest_score = 15 x_val = 4.45 else: unrest_score = 15 x_val = 4.45 else: if self.reportLanguage == "en": if self.unrest < 5: x_val = self.unrest / 5 * 0.95 unrest_score = 0 elif (5 <= self.unrest) and (self.unrest < 10): x_val = (self.unrest - 5) / 5 * 0.95 + 0.95 unrest_score = 3 elif (10 <= self.unrest) and (self.unrest < 15): x_val = (self.unrest - 10) / 4 * 0.95 + 1.9 unrest_score = 6 elif (15 <= self.unrest) and (self.unrest < 20): x_val = (self.unrest - 15) / 5 * 0.95 + 2.95 unrest_score = 10 elif (20 <= self.unrest) and (self.unrest < 28): x_val = (self.unrest - 20) / 8 * 1 + 3.9 unrest_score = 15 else: x_val = (self.unrest - 20) / 8 * 1 + 3.9 unrest_score = 15 else: if self.unrest < 10: x_val = self.unrest / 10 * 0.95 unrest_score = 0 elif (10 <= self.unrest) and (self.unrest < 15): x_val = (self.unrest - 10) / 5 * 0.95 + 0.95 unrest_score = 3 elif (15 <= self.unrest) and (self.unrest < 20): x_val = (self.unrest - 15) / 5 * 0.95 + 1.9 unrest_score = 6 elif (20 <= self.unrest) and (self.unrest < 30): x_val = (self.unrest - 20) / 10 * 0.95 + 2.95 unrest_score = 10 elif (30 <= self.unrest) and (self.unrest < 46): x_val = (self.unrest - 30) / 17 * 1 + 3.9 unrest_score = 15 else: x_val = (self.unrest - 30) / 17 * 1 + 3.9 unrest_score = 15 self.report['anx'] = round(self.unrest,2) return x_val, unrest_score ''' 주관적 심리상태 계산 ''' def cal_pse(self): print("Psychological State Evaluation") print("depress:", self.depressed, "anx:", self.unrest) x_val_dep, self.depressed_score = self.cal_depress(reportLanguage=self.reportLanguage) x_val_anx, self.unrest_score = self.cal_unrest(reportLanguage=self.reportLanguage) if int(self.real_age) < 14: PLOT.plot_pse_14(x_val_dep, 'depress', self.RESULT_DIR) PLOT.plot_pse_14(x_val_anx, 'unrest', self.RESULT_DIR) else: PLOT.plot_pse(x_val_dep, 'depress', self.RESULT_DIR) PLOT.plot_pse(x_val_anx, 'unrest', self.RESULT_DIR) def cal_ppg_score(self): score_ans = np.array(self.zscore_ans) - 1 # 정상범위 상단 밑에있는 항목들만 z-score차감. score_ans[(score_ans >= 0)] = 0 score_ab = (10 - abs(self.sns - 50) * 0.2) * 1.0 # 10 point / 중간일수록 점수 높게 score_si = ((3 - self.zscore_si)/6 * 10) * 1.0 # 10 point / 높을수록 점수 낮게 score_sr = ((3 + self.zscore_sr)/6 * 10) * 1.0 # 10 point / 높을수록 점수 높게 score_si = np.clip(score_si, 0, 10) score_sr = np.clip(score_sr, 0, 10) score_list = np.array([15 - abs(score_ans).sum() * 1.5, score_ab, score_si, score_sr, 15 - self.depressed_score, 15 - self.unrest_score]) # ans:15, ab:10, si:10, sr:10, depress:15, unrest:15 score_list = score_list.clip(0) # 설문지 제외 HRV ONLY SCORE : hrv항목 4개 합 * 70/45 + 35 self.hrv_only_score = np.sum(score_list[:4]) * (70/45) + 35 score = score_list.sum() + 25 if score > 99: score = 99 self.ppg_score = score # ppg_score self.report['ppgScore'] = round(self.ppg_score, 2) # ppg_hrv_only self.report['ppgOnlyScore'] = round(self.hrv_only_score, 2) PLOT.plot_score(round(self.ppg_score), 'ppg', self.RESULT_DIR, reportLanguage=self.reportLanguage) print("ppg_score:", self.ppg_score) return score_list def count_score(self, li): big = [i for i in li if 84.13 <= float(i)] small = [i for i in li if 15.86 >= float(i)] return [len(big), len(small)] def cal_emoji(self): ppg_emoji = [0, 0, 0, 0, 0] # ANS ratio_ans = st.norm.cdf(self.zscore_ans) * 100 count = self.count_score(ratio_ans) if count[1] >= 2: ppg_emoji[0] = 5 elif count[1] == 1: ppg_emoji[0] = 4 elif count[0] == 0 & count[1] == 0: ppg_emoji[0] = 3 elif count[0] == 1: ppg_emoji[0] = 2 elif count[0] >= 2: ppg_emoji[0] = 1 # AB if 40 < self.sns < 60: ppg_emoji[1] = 1 elif 60 <= self.sns <= 80: ppg_emoji[1] = 2 elif 20 <= self.sns <= 40: ppg_emoji[1] = 3 elif 80 < self.sns: ppg_emoji[1] = 4 else: ppg_emoji[1] = 5 # Stress ratio_si = st.norm.cdf(self.zscore_si) * 100 ratio_sr = st.norm.cdf(self.zscore_sr) * 100 ppg_emoji[2] = 3 print("ratio_si", ratio_si, "ratio sr", ratio_sr) if ratio_si < 33.33 and ratio_sr > 50: ppg_emoji[2] = 1 if ratio_si < 50 and ratio_sr > 66.66: ppg_emoji[2] = 1 if (33.33 <= ratio_si <= 66.66) and (33.33 <= ratio_sr <= 66.66): ppg_emoji[2] = 3 if ratio_si > 50 and ratio_sr > 66.66: ppg_emoji[2] = 2 if ratio_si < 50 and ratio_sr < 33.33: ppg_emoji[2] = 4 if ratio_si > 66.66 and ratio_sr < 50: ppg_emoji[2] = 5 # PSE if self.depressed == 0 and self.unrest == 0: ppg_emoji[3] = 1 elif self.depressed_score < 3 and self.unrest_score < 3: ppg_emoji[3] = 2 else: if self.depressed_score >= 3 or self.unrest_score >= 3: ppg_emoji[3] = 3 if self.depressed_score >= 6 or self.unrest_score >= 6: ppg_emoji[3] = 4 if self.depressed_score >= 10 or self.unrest_score >= 10: ppg_emoji[3] = 5 self.ppg_emoji = ppg_emoji ''' 자기 이해도 계산 ''' def cal_self_understanding(self, score_list): # ppg점수(pse점수제외)와 pse 비교 score_mean = (self.ppg_score - score_list[4] - score_list[5] - 25) / 3 du_mean = (score_list[4] + score_list[5]) / 2 if du_mean < 2: du_mean = 2 self_u = abs(round((du_mean / score_mean) * 100, 1)) self_u = 100 - abs(self_u - 100) if self_u > 100: self_u = self_u - 100 if self_u < 10: self_u = 10 if (score_list[4] <= 9 or score_list[5] <= 9) and (score_mean <= 6.5) and (self_u <= 33.33): self_u = 33.33 self.self_u = round(self_u,2) # Self Understanding if self.self_u >= 66.66: self.ppg_emoji[4] = 1 elif (self.self_u >= 33.33) and (self.self_u < 66.66): self.ppg_emoji[4] = 3 else: self.ppg_emoji[4] = 5 self.report['ppg_emoji']=self.ppg_emoji # plot self.report['self_u']=round(self_u) PLOT.plot_self_understanding(self_u, self.RESULT_DIR) def cal_ppg_ebsi(self): ratio_ans = st.norm.cdf(self.zscore_ans) * 100 count = self.count_score(ratio_ans) self.ppg_ebsi_value = [100 - ratio_ans[0], 100 - self.sns, self.sns] self.ppg_ebsi = directory.mind_ebsi_index["안정형"] if count[1] >= 2 and self.sns > 75: self.ppg_ebsi = directory.mind_ebsi_index["심약불안형"] elif count[1] >= 2 and self.sns < 25: self.ppg_ebsi = directory.mind_ebsi_index["심약우울형"] else: if count[1] >= 2: self.ppg_ebsi = directory.mind_ebsi_index["심약형"] if self.sns > 75: self.ppg_ebsi = directory.mind_ebsi_index["불안형"] if self.sns < 25: self.ppg_ebsi = directory.mind_ebsi_index["우울형"] if count[1] >= 2: self.ppg_ebsi_value[0] = 100 - (np.mean(sorted(ratio_ans)[:2])) self.ppg_ebsi_detail = [np.random.randint(1, directory.mind_ebsi_length["".join([str(self.ppg_ebsi), "_1"])]+1), np.random.randint(1, directory.mind_ebsi_length["".join([str(self.ppg_ebsi), "_2"])]+1), np.random.randint(1, directory.mind_ebsi_length["".join([str(self.ppg_ebsi), "_3"])]+1)] if int(self.real_age) < 20: # 미성년자일 경우 ppg_ebsi_detail 어린이것만 가져오도록 수정 if self.ppg_ebsi == 1: self.ppg_ebsi_detail[1] = np.random.choice([1,3,4]) elif self.ppg_ebsi == 3: self.ppg_ebsi_detail[2] = np.random.choice([1,3]) if (self.ppg_ebsi == 6) and (int(self.ppg_emoji[2]) == 5): self.ppg_ebsi_detail[0] = np.random.randint(2, 3+1) elif (self.ppg_ebsi == 6) and (int(self.ppg_emoji[3]) == 5): self.ppg_ebsi_detail[0] = np.random.randint(4, 4+1) else: self.ppg_ebsi_detail[0] = np.random.randint(1, 1+1) PLOT.plot_PPG_ebsi_radar(self.ppg_ebsi_value, self.RESULT_DIR) self.report['ppg_ebsi_value'] = [round(v,2) for v in self.ppg_ebsi_value] self.report['ppg_subtype'] = [k for k, v in directory.mind_ebsi_index.items() if (v == self.ppg_ebsi) and (k.upper() == k.lower())][0] # value to key & 한국어 가져오기 def ppg_dq(self): # 웹에서 ppg_dq 를 먼저 받아오기 위해 함수 만듬 self.load_file() self.preprocess() self.cal_time_freq_domain() return self.ppg_dq def cal_stats_ppg(self): age = directory.age_modify(self.age, png=True) info = pd.read_csv(os.path.join(directory.ROOT_DIR, 'all_info.csv')) info.index = ['eeg_mean', 'eeg_std', 'ppg_mean', 'ppg_std'] age_ppg_mean = info[age]['ppg_mean'] age_ppg_std = info[age]['ppg_std'] ppg_zscore = (self.ppg_score - age_ppg_mean) / age_ppg_std percent = round(100 - st.norm.cdf(ppg_zscore) * 100) self.report['ppg_norm_percent'] = percent def run_function_ppg(self): #start_time = time.perf_counter() # dq 시간 확인 self.load_file() self.preprocess() self.cal_time_freq_domain() #end_time = time.perf_counter() #print(f"PPG DQ time elapsed : {int(round((end_time - start_time) * 1000))}ms") self.cal_ans_activity() self.cal_ans_balance() self.cal_stress_index() self.cal_stress_resistance() self.cal_pse() self.cal_emoji() self.cal_ppg_ebsi() score_list = self.cal_ppg_score() self.cal_self_understanding(score_list) self.cal_stats_ppg() score = {'filename': self.file.split('/')[-1][:-4], 'ppg': round(self.ppg_score, 1), 'ppg_hrv_only': round(self.hrv_only_score,1), 'ans': round(score_list[0], 1), 'ab': round(score_list[1], 1), 'si': round(score_list[2], 1), 'sr': round(score_list[3], 1), 'depress': round(score_list[4], 1), 'unrest': round(score_list[5], 1), 'ppg_emoji': str(self.ppg_emoji).split('[')[1].split(']')[0], 'ppg_dq': self.data_quality_ppg, 'ppg_ebsi': self.ppg_ebsi, 'ppg_ebsi_detail': str(self.ppg_ebsi_detail).split('[')[1].split(']')[0], 'hrt': round(self.hrt)} print(score) result = pd.DataFrame(score, index=[0]) result['ppg_ebsi_value'] = '' result.at[result.index[0], 'ppg_ebsi_value'] = [round(i, 1) for i in self.ppg_ebsi_value] result.to_csv(os.path.join(self.RESULT_DIR, self.file.split('/')[-1][:-4] + '_ppg.csv'), index=False, encoding='utf-8-sig')