import binascii import os import mne import numpy as np import pandas as pd import peakutils import scipy.stats as st from matplotlib import pyplot as plt from scipy import signal import PLOT import time import re import directory from directory import FREQ_BANDS_alpha from directory import N_DATA_EEG import json import ast class EEG: def __init__(self, file, RESULT_DIR, age='20', device='laxtha', reportLanguage="kr"): self.file = file self.RESULT_DIR = RESULT_DIR self.device = device self.samplefreq = directory.cal_samplefreq(self.file, self.device) # soso24 구분을 위해 file 인수 추가 self.real_age = age # 웹에서 실제 받아오는 age self.age = directory.age_modify(age) # z-score 계산을 위한 연령대 그룹으로 자름 self.pre_epochs = None self.epochs = None self.psds = None self.freqs = None self.psds_mean = None # all channel psds mean self.indices = None self.peak_x = None self.peak_y = None self.ch1_rel = None self.ch2_rel = None self.fas = None self.com = None self.zscore_psd = None self.zscore_apf_freq = None self.zscore_apf_psd = None self.zscore_apf_yn = 0 self.zscore_peak_x_yn = 0 self.zscore_rel = None self.zscore_fas = None self.zscore_com = None self.eeg_score = None self.epochs_ratio = None self.data_quality_eeg = "high" self.eeg_emoji = None self.eeg_ebsi = None self.epoch_new = None self.epoch_old = None self.reportLanguage = reportLanguage if self.reportLanguage == "en": PLOT.BASE_DIR = directory.BASE_DIR_EN self.report = {'eeg_code': -2005,'eegScore': 0, 'dataQuality_eeg': 'bad', 'eeg_subtype': None} # 레포트 출력을 위한 결과값 self.viewer_data = {'eeg_ch1': [], 'eeg_ch2': []} self.dq = {'dataQuality_eeg':'bad'} def isNaN(self, num): if num != num: return "none" else: return "desc" ''' 파일 읽기 (각 기기마다 raw file 확장자 및 load 하는 방식이 다름) ''' def load_file(self, tmax=305): _, 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: ch1 = i[4:10] ch2 = i[10:16] ch1_trans = directory.transHex2Deci(ch1[0:2], ch1[2:4], ch1[4:6]) ch2_trans = directory.transHex2Deci(ch2[0:2], ch2[2:4], ch2[4:6]) line = ' '.join([str(ch1_trans), str(ch2_trans)]) temp.append(line) f.close() data = [temp[i].split() for i in range(0, len(temp))] if ((self.device == "soso") or (self.device == "laxtha") or (self.device == "laxtha_old")) and (extension == ".txt"): f = open(self.file, "r", encoding='euc-kr') # open txt file, UTF-8 encoding temp = [] while True: line = f.readline() if not line: break temp.append(line) f.close() # device unmatch 검증 data = [temp[i].split()[0:2] for i in range(7, len(temp))] # header 제외 if (self.device == "crown") and (extension == ".txt"): temp = [] ch_names = ['CP3', 'C3', 'F5', 'PO3', 'PO4', 'F6', 'C4', 'CP4'] csv_header = ['startTime'] + ch_names f = open(self.file, "r") lst = [p.strip() for p in f.read().split('\n\n')] for event in range(len(lst)): sample_epoch = [[x] for x in lst[event].split('\n')] temp.append([[round(eval(x), 3) for x in sample_epoch[ch][0].split(',')] for ch in range(1, len(csv_header))]) f.close() data = np.array(temp) data = data.transpose(1, 0, 2) data = data.reshape(data.shape[0], data.shape[1] * data.shape[2]) # shape: 8 X number of samples data = np.transpose(data) data = data[:, [2, 5]] if (self.device == "crown") and (extension == ".csv"): temp = [] df = pd.read_csv(self.file) df = df[['F5', 'F6']] data = df.values if (self.device == "cgx") and (extension == ".txt"): ch_names = ['F7', 'Fp1', 'Fp2', 'F8', 'F3', 'Fz', 'F4', 'C3', 'Cz', 'P8', 'P7', 'Pz', 'P4', 'T3', 'P3', 'O1', 'O2', 'C4', 'T4'] data_df = pd.read_table(self.file, skiprows=[0, 1, 2, 3], sep="\t", low_memory=False, on_bad_lines='skip') cols = list(np.arange(1, 20)) data_df = data_df[data_df.columns[cols]] data_df.replace(',', '', regex=True, inplace=True) for col in range(len(ch_names)): if data_df[data_df.columns[col]].dtype != 'float64': data_df[data_df.columns[col]] = pd.to_numeric(data_df[data_df.columns[col]]) data = data_df.values[:, 1:3] 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버전 탐 #print('soso_24 version') # 넘어오는거 확인 용으로 print(라이브 동기화할 땐 안해도 됨 ) f = open(self.file, "rb") t = f.read() by = re.findall('a0\w{62}c0', str(t)) temp = [] for i in by: ch1 = i[4:10] ch2 = i[10:16] ch1_trans = directory.transHex2Deci_24(ch1[0:2], ch1[2:4], ch1[4:6]) ch2_trans = directory.transHex2Deci_24(ch2[0:2], ch2[2:4], ch2[4:6]) line = ' '.join([str(ch1_trans), str(ch2_trans)]) temp.append(line) f.close() else: #print("soso 19/23") f = open(self.file, "rb") t = f.read() by = re.findall('a0\w{62}c0', str(t)) temp = [] for i in by: ch1 = i[4:10] ch2 = i[10:16] ch1_trans = directory.transHex2Deci(ch1[0:2], ch1[2:4], ch1[4:6]) ch2_trans = directory.transHex2Deci(ch2[0:2], ch2[2:4], ch2[4:6]) line = ' '.join([str(ch1_trans), str(ch2_trans)]) temp.append(line) f.close() data = [temp[i].split() for i in range(0, len(temp))] if self.device == 'linkband': #룩시드랩스 기기 with open(self.file, 'r', encoding='utf-8') as f: js = json.load(f) # 키 후보: 'eegChannel1/2' 또는 'ch1/ch2' ch1 = js.get('eegChannel1', js.get('ch1', [])) ch2 = js.get('eegChannel2', js.get('ch2', [])) 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(ch1, str): ch1 = ast.literal_eval(ch1) if isinstance(ch2, str): ch2 = ast.literal_eval(ch2) # numpy 변환 ch1 = np.asarray(ch1, dtype=float) ch2 = np.asarray(ch2, dtype=float) # 길이 맞추기 (두 채널 공통 최소 길이) n = min(len(ch1), len(ch2)) ch1, ch2 = ch1[:n], ch2[:n] # (선택) µV 스케일 필요 시 주석 해제 #Vref, gain, res = 4.033, 12.0, 8388607.0 #ch1 = (ch1 / res) * (Vref / gain) * 1e6 #ch2 = (ch2 / res) * (Vref / gain) * 1e6 df = pd.DataFrame({'ch1': ch1, 'ch2': ch2}) data = df.values data = pd.DataFrame(data, columns=['Fp1', 'Fp2']) data = data.dropna() data = data.astype('float') # * 1e-6 if (self.device == "light_soso") and (extension == ".txt") and (self.file.split('/')[-1].split('_')[1] =='01'): # light_soso eeg 불량 처리위해 기준 추가 24.07.25 data = data.loc[(abs(data['Fp1']) < 1000) & (abs(data['Fp2']) < 1000)] info = mne.create_info(ch_names=['Fp1', 'Fp2'], sfreq=self.samplefreq, ch_types='eeg') rawfile = mne.io.RawArray(np.transpose(data.to_numpy()), info, verbose=False) if rawfile.times[-1] > tmax: rawfile = rawfile.crop(tmax=tmax) self.raw = rawfile rawfile = rawfile.notch_filter(freqs=np.arange(60, 121, 60)) iir_params = dict(order=6, ftype='butter') filtered = rawfile.filter(4, 41, n_jobs=1, verbose=False, method='iir', iir_params=iir_params) pre_epochs = mne.make_fixed_length_epochs(filtered, duration=2, preload=False, verbose=False).load_data() self.pre_epochs = pre_epochs def viewr_preprocess(self): raw = self.raw.copy().get_data() ch1 = raw[0, :] ch2 = raw[1, :] eeg_x = np.arange(0, len(ch1)/self.samplefreq, 1/self.samplefreq) ch1_df = pd.DataFrame(np.array([eeg_x, ch1]).T, columns=['time', 'eeg1']) ch2_df = pd.DataFrame(np.array([eeg_x, ch2]).T, columns=['time', 'eeg2']) # result.to_csv(os.path.join(self.RESULT_DIR, self.file.split('/')[-1][:-4]+'_eeg.csv'), index=False, encoding='utf-8-sig') return ch1_df, ch2_df ''' EEG 전처리 ''' def preprocess(self): # Global Threshold pre_e_dat = self.pre_epochs.get_data() epoch_no, _, _ = pre_e_dat.shape epoch_old = epoch_no self.epoch_old = epoch_old if self.device == 'laxtha': e_index = [e for e in range(epoch_no) if pre_e_dat[e, :, :].max() > 75 or pre_e_dat[e, :, :].min() < -75] else: e_index = [e for e in range(epoch_no) if pre_e_dat[e, :, :].max() > 100 or pre_e_dat[e, :, :].min() < -100] self.pre_epochs.drop(e_index, verbose=False) # Individual Threshold pre_e_dat = self.pre_epochs.get_data() bad_t_index = [] for p in range(pre_e_dat[0, :, 0].shape[0]): maxt = np.max(abs(pre_e_dat[:, p]), axis=1) threshold = np.mean(maxt) + 3 * np.std(maxt) for q in range(len(maxt)): if maxt[q] > threshold: bad_t_index.append(q) bad_t_d_index = [] # 차분 for p in range(pre_e_dat[0, :, 0].shape[0]): maxt_d = np.max(abs(np.diff(pre_e_dat[:, p])), axis=1) threshold = np.mean(maxt_d) + 3 * np.std(maxt_d) for q in range(len(maxt_d)): if maxt_d[q] > threshold: bad_t_d_index.append(q) bad_e_index = bad_t_index + bad_t_d_index bad_e_index = sorted(list(set(bad_e_index))) self.pre_epochs.drop(bad_e_index, verbose=False) epochs = self.pre_epochs # PSD Threshold epoch_dat = epochs.get_data() epoch_no, _, _ = epoch_dat.shape psds, freqs = mne.time_frequency.psd_welch(epochs, fmin=4., fmax=40., n_fft=2*self.samplefreq, verbose=False, n_jobs=1, n_overlap = 1*self.samplefreq-1, n_per_seg=int(1*self.samplefreq), window=signal.windows.hann(int(1*self.samplefreq))) # Hanning window bad_p_index = [] for p in range(psds[0, :, 0].shape[0]): ch = psds[:, p, :] for q in range(len(ch[0])): threshold = np.mean(ch[:, q]) + 5 * (np.std(ch[:, q])) for r in range(len(ch[:, 0])): if ch[r, q] > threshold: bad_p_index.append(r) bad_p_index = sorted(list(set(bad_p_index))) epochs.drop(bad_p_index, verbose=False) epoch_new = epochs.get_data().shape[0] self.epochs = epochs psds, freqs = mne.time_frequency.psd_welch(self.epochs, fmin=4., fmax=40., n_fft=2*self.samplefreq, verbose=False, n_jobs=1, n_overlap = 1*self.samplefreq-1, n_per_seg=int(1*self.samplefreq), window=signal.windows.hann(int(1*self.samplefreq))) # Hanning window bad_p_index = [] psds /= np.sum(psds, axis=-1, keepdims=True) # absolute power to relative power self.psds = psds self.freqs = freqs self.epochs_ratio = epoch_new / epoch_old self.epoch_new = epoch_new self.epoch_old = epoch_old if self.epoch_old > 85: #170초 이상 파일에서는 전처리 후 90초 미만일 경우 오류 발생 assert self.epoch_new >= 45, "This data too short after preprocessing, under 90sec" else: #이외의 경우(1분 측정) 전처리 후 30초 미만일 경우 오류 발생 assert self.epoch_new >= 15, "This data too short after preprocessing, under 30sec" return epochs, psds, freqs ''' Alpha peak 찾음 -> 없으면 cal_cog 로 두뇌 총명도 계산 ''' def detect_peak(self, x): indices = peakutils.indexes(self.psds_mean[6:22]) # detected peak index if len(indices) == 0: indices = [] peak_x = [] peak_y = [] elif indices[-1] == 1: indices = peakutils.indexes(self.psds_mean[6:22]) peak_x = x[6:22] peak_y = self.psds_mean[x.index(x[6:22][indices[-1]])] elif indices[-1] == 2: indices = peakutils.indexes(self.psds_mean[7:21]) peak_x = x[7:21] peak_y = self.psds_mean[x.index(x[7:21][indices[-1]])] else: indices = peakutils.indexes(self.psds_mean[8:22]) peak_x = x[8:22] peak_y = self.psds_mean[x.index(x[8:22][indices[-1]])] return indices, peak_x, peak_y ''' 두뇌총명도 계산1 ''' def cal_psd(self): self.psds_mean = self.psds.mean(axis=0).mean(axis=0) normal_psds_ch_avg = N_DATA_EEG[''.join(['psds_ch_', self.age, '_avg'])].values normal_psds_ch_std = N_DATA_EEG[''.join(['psds_ch_', self.age, '_std'])].values zscore_psd = (self.psds_mean - normal_psds_ch_avg) / normal_psds_ch_std x = [i / 2 + 4 for i in range(len(self.freqs))] self.indices, self.peak_x, self.peak_y = self.detect_peak(x) if len(self.indices) > 0: self.peak_x = self.peak_x[self.indices[-1]] self.zscore_psd = zscore_psd.mean() ''' 두뇌 총명도 계산2 ''' def cal_apf(self): normal_apf_freq_avg = N_DATA_EEG[''.join(['apf_freq_', self.age, '_avg'])].dropna().values normal_apf_freq_std = N_DATA_EEG[''.join(['apf_freq_', self.age, '_std'])].dropna().values normal_apf_psd_avg = N_DATA_EEG[''.join(['apf_psd_', self.age, '_avg'])].dropna().values normal_apf_psd_std = N_DATA_EEG[''.join(['apf_psd_', self.age, '_std'])].dropna().values self.zscore_apf_freq = (self.peak_x - normal_apf_freq_avg) / normal_apf_freq_std self.zscore_apf_psd = (np.sqrt(self.peak_y) - normal_apf_psd_avg) / normal_apf_psd_std # plot apf_freq = self.peak_x apf_psd = st.norm.cdf(self.zscore_apf_psd[0]) * 100 if apf_freq < 8: apf_freq = 8 if apf_freq > 12: apf_freq = 12 PLOT.plot_brain_intelligence(apf_freq, apf_psd, self.RESULT_DIR) self.report['apf_freq'] = round(apf_freq,1) self.report['apf_psd'] = round(apf_psd,1) def cal_cog(self): normal_apf_freq_avg = N_DATA_EEG[''.join(['apf_freq_', self.age, '_avg'])].dropna().values normal_apf_freq_std = N_DATA_EEG[''.join(['apf_freq_', self.age, '_std'])].dropna().values normal_apf_psd_avg = N_DATA_EEG[''.join(['apf_psd_', self.age, '_avg'])].dropna().values normal_apf_psd_std = N_DATA_EEG[''.join(['apf_psd_', self.age, '_std'])].dropna().values x = [i / 2 + 4 for i in range(len(self.freqs))] peak_x = (x[10:19] * self.psds_mean[10:19]).sum() / self.psds_mean[10:19].sum() self.peak_x = round(peak_x / .5) * .5 self.peak_y = self.psds_mean[x.index(self.peak_x)] self.zscore_apf_freq = (self.peak_x - normal_apf_freq_avg) / normal_apf_freq_std self.zscore_apf_psd = (np.sqrt(self.peak_y) - normal_apf_psd_avg) / normal_apf_psd_std # plot apf_freq = 8.5 apf_psd = 10 PLOT.plot_brain_intelligence(apf_freq, apf_psd, self.RESULT_DIR) ''' 두뇌 활성도 계산 ''' def cal_relative_power(self): normal_rel_avg = N_DATA_EEG[''.join(['rel_ch_'+self.age+'_avg'])].dropna().values normal_rel_std = N_DATA_EEG[''.join(['rel_ch_'+self.age+'_std'])].dropna().values temp = [] for fmin, fmax in FREQ_BANDS_alpha.values(): psds_band = self.psds[:, 0, (self.freqs >= fmin) & (self.freqs < fmax)].mean(axis=0).sum() temp.append(psds_band) self.ch1_rel = [round(temp[i] / (sum(temp) - temp[1]) * 100, 4) for i in range(len(temp))] temp = [] for fmin, fmax in FREQ_BANDS_alpha.values(): psds_band = self.psds[:, 1, (self.freqs >= fmin) & (self.freqs < fmax)].mean(axis=0).sum() temp.append(psds_band) self.ch2_rel = [round(temp[i] / (sum(temp) - temp[1]) * 100, 4) for i in range(len(temp))] ch_rel = [round((self.ch1_rel[i] + self.ch2_rel[i]) / 2, 4) for i in range(len(self.ch1_rel))] zscore_rel = (ch_rel - normal_rel_avg) / normal_rel_std ratio_rel = np.delete(st.norm.cdf(zscore_rel) * 100, 1) zscore_rel = np.delete(zscore_rel, 1) self.zscore_rel = zscore_rel # plot PLOT.plot_brain_activity(ratio_rel, self.RESULT_DIR) self.report['score_rel'] = [int(v) for v in ratio_rel] return ch_rel ''' 두뇌 균형도 계산 ''' def cal_faa(self): normal_fas_avg = N_DATA_EEG['fas_ch_avg'].dropna().values normal_fas_std = N_DATA_EEG['fas_ch_std'].dropna().values al = self.ch1_rel[1] ar = self.ch2_rel[1] self.fas = np.log(al) - np.log(ar) # relative log transform formula zscore_fas = (self.fas - normal_fas_avg[0]) / normal_fas_std[0] self.zscore_fas = zscore_fas ratio_fas = st.norm.cdf(zscore_fas) * 100 if ratio_fas < 5: ratio_fas = 5 if ratio_fas > 95: ratio_fas = 95 # plot PLOT.plot_brain_balance(ratio_fas, self.RESULT_DIR, reportLanguage=self.reportLanguage) self.report['ratio_fas'] = ratio_fas return self.fas def kolmogorov_complexity(self, arr): mean = arr.mean() for i in range(len(arr)): if arr[i] >= mean: arr[i] = 1 else: arr[i] = 0 n = len(arr) c = 1 l = 1 i = 0 k = 1 k_max = 1 stop = 0 while stop == 0: if arr[i + k - 1] != arr[l + k - 1]: if k > k_max: k_max = k i = i + 1 if i == l: c = c + 1 l = l + k_max if l + 1 > n: stop = 1 else: i = 0 k = 1 k_max = 1 else: k = 1 else: k = k + 1 if l + k > n: c = c + 1 stop = 1 b = n / np.log2(n) complexity = c / b return complexity ''' 두뇌 유연성 계산 ''' def cal_complexity(self): epoc_dat = self.epochs.get_data() _, sen_no, _ = epoc_dat.shape comp_mat = [] for i in range(sen_no): arr = epoc_dat[:, i, :] temp_comp = [] for j in range(arr.shape[0]): arr_1 = arr[j, :] k = directory.kolmogorov_complexity(arr_1) temp_comp.append(k) comp_mat.append(np.mean(temp_comp)) self.com = np.mean(comp_mat) normal_com_avg = N_DATA_EEG[''.join(['com_ch_', self.age, '_avg'])].dropna().values normal_com_std = N_DATA_EEG[''.join(['com_ch_', self.age, '_std'])].dropna().values self.zscore_com = (self.com - normal_com_avg[0]) / normal_com_std[0] ratio_com = st.norm.cdf(self.zscore_com) * 100 if (7 <= self.peak_x <= 8.5) and (st.norm.cdf(self.zscore_apf_psd) > .75): ratio_com = 32 self.zscore_com = st.norm.ppf(.32) if (self.zscore_apf_psd > 1) and (ratio_com < 32): ratio_com = 32 self.zscore_com = st.norm.ppf(.32) # plot PLOT.plot_brain_flexibility(ratio_com, self.RESULT_DIR) self.report['ratio_com'] = round(ratio_com,2) return self.com def cal_eeg_score(self): temp_zscore_rel = self.zscore_rel.copy() theta_penalty = 0 temp_theta = temp_zscore_rel[0] if self.eeg_ebsi in [directory.brain_ebsi_index["안개긴장형"], directory.brain_ebsi_index["안개형"]]: theta_penalty = 3 if temp_zscore_rel[1] > 0: #if Low alpha > norm, no deduction for point temp_zscore_rel[1] = 0 if temp_zscore_rel[2] > 0: #if High alpha > norm, no deduction for point temp_zscore_rel[2] = 0 score_rel = np.array([0 if abs(i) < 1 else i for i in temp_zscore_rel[1:]]) #if any power in norm range, no deduction for point score_rel = np.insert(score_rel, 0, abs(temp_theta)) score_rel = 20 - abs(score_rel).sum()*1.5 score_com = 10 - (abs(st.norm.cdf(self.zscore_com)*100-50) * 0.2) score_faa = 10 - (abs(st.norm.cdf(self.zscore_fas)*100-50) * 0.2) score_apf_freq = st.norm.cdf(self.zscore_apf_freq)*100 * 0.05 score_apf_psd = st.norm.cdf(self.zscore_apf_psd)*100 * 0.1 score_list = np.array([score_rel, score_com, score_faa, score_apf_freq[0], score_apf_psd[0]]) # rel: 20, com: 10, faa: 10, apf_freq: 5, apf_psd: 10 GV score_list = score_list.clip(0) score = score_list.sum() + 45 score = score - self.zscore_apf_yn - self.zscore_peak_x_yn - theta_penalty if score > 99: score = 99 self.eeg_score = score # plot PLOT.plot_score(round(self.eeg_score), 'eeg', self.RESULT_DIR, reportLanguage=self.reportLanguage) self.report['eegScore'] = round(self.eeg_score, 2) return score_list def count_zscore(self, list): filtered = [i for i in list if (-1 > float(i)) or (float(i)) > 1] return len(filtered) ''' 두뇌 report 내 각 항목별 emoji 설정 ''' def cal_emoji(self): # Priority: if theta > norm -> dangerous eeg_emoji = [0, 0, 0, 0] # Brain Activity if (self.eeg_ebsi == directory.brain_ebsi_index["안개형"]) or (self.eeg_ebsi == directory.brain_ebsi_index["안개긴장형"]): eeg_emoji[0] = 5 elif self.zscore_rel[3] >= 1 or self.zscore_rel[4] >= 1 or self.zscore_rel[5] >= 1 or self.zscore_rel[6] >= 1: eeg_emoji[0] = 4 else: if self.count_zscore(self.zscore_rel) == 0: eeg_emoji[0] = 1 elif self.count_zscore(self.zscore_rel) <= 2: eeg_emoji[0] = 2 else: eeg_emoji[0] = 3 # Complexity ratio_com = st.norm.cdf(self.zscore_com) * 100 if ratio_com < 15.86: eeg_emoji[1] = 41 elif ratio_com > 84.13: eeg_emoji[1] = 42 else: if 15.86 <= ratio_com <= 33.33: eeg_emoji[1] = 31 elif 66.66 <= ratio_com <= 84.13: eeg_emoji[1] = 32 else: eeg_emoji[1] = 2 # APF if self.zscore_peak_x_yn != 0 or self.zscore_apf_yn != 0: eeg_emoji[2] = 5 else: x1 = self.peak_x if x1 < 8: x1 = 8 if x1 > 12: x1 = 12 y1 = -25 * x1 + 250 y2 = -25 * x1 + 275 y3 = -25 * x1 + 325 y4 = -25 * x1 + 350 temp_psd = st.norm.cdf(self.zscore_apf_psd[0])*100 if x1 >= 10: if temp_psd >= y4: eeg_emoji[2] = 1 elif temp_psd >= y3: eeg_emoji[2] = 2 elif temp_psd >= y2: eeg_emoji[2] = 3 else: eeg_emoji[2] = 4 else: if temp_psd < y1: eeg_emoji[2] = 5 elif temp_psd < y2: eeg_emoji[2] = 4 elif temp_psd < y3: eeg_emoji[2] = 3 else: eeg_emoji[2] = 2 # FAA ratio_faa = st.norm.cdf(self.zscore_fas) * 100 if 33.33 < ratio_faa < 66.66: eeg_emoji[3] = 1 if 15.86 < ratio_faa <= 33.33: eeg_emoji[3] = 2 if 66.66 <= ratio_faa < 84.13: # 1sigma eeg_emoji[3] = 3 if ratio_faa <= 15.86: eeg_emoji[3] = 4 if ratio_faa >= 84.13: eeg_emoji[3] = 5 self.eeg_emoji = eeg_emoji self.report['eeg_emoji'] = self.eeg_emoji def cal_data_quality(self): if self.epochs_ratio >= 0.7:#기존 0.75에서 0.85로 변경 25.01.13 -> 0.7로 변경 25.05.30 self.data_quality_eeg = "high" elif self.epochs_ratio >= 0.6:#기존 0.55에서 0.7로 변경 25.01.13 -> 0.6로 변경 25.05.30 self.data_quality_eeg = "middle" else: self.data_quality_eeg = "low" #if self.epoch_new < 45: #self.data_quality_eeg = "low" #print("This data Too short < 90sec") 1분,3분 도입으로 인해 90초이하 품질 low 주던거 제외 self.report['dataQuality_eeg'] = self.data_quality_eeg self.dq['dataQuality_eeg'] = self.data_quality_eeg def cal_ebsi_eeg(self): ratio_rel = st.norm.cdf(self.zscore_rel) * 100 highF = self.zscore_rel[3:7] # low beta, middle beta, high beta, gamma highF_ratio = st.norm.cdf(highF) * 100 highF_max = np.max(highF) zscore_rel_max = np.max(self.zscore_rel) self.eeg_ebsi_value = [ratio_rel[0], ratio_rel[2], np.mean(highF_ratio), ratio_rel[1]] self.eeg_ebsi = directory.brain_ebsi_index["안정형"] if (self.zscore_rel[0] >= 1 and self.eeg_emoji[2] == 5) and highF_max >= 1:#기존 총명도50프로 아래 ->총명도 위험일 때만 안개긴장형 self.eeg_ebsi = directory.brain_ebsi_index["안개긴장형"] elif (self.zscore_rel[0] >= 1 and self.eeg_emoji[2] == 5):#기존 총명도50프로 아래 ->총명도 위험일 때만 안개형 25.01.13 self.eeg_ebsi = directory.brain_ebsi_index["안개형"] else: if self.zscore_rel[2] >= 1 and zscore_rel_max == self.zscore_rel[2]: self.eeg_ebsi = directory.brain_ebsi_index["영재형"] elif self.zscore_rel[1] >= 1 and zscore_rel_max == self.zscore_rel[1]: self.eeg_ebsi = directory.brain_ebsi_index["차분형"] elif highF_max >= 1 and zscore_rel_max == highF_max: self.eeg_ebsi = directory.brain_ebsi_index["긴장형"] if self.eeg_ebsi in [directory.brain_ebsi_index["안개긴장형"], directory.brain_ebsi_index["긴장형"]]: self.eeg_ebsi_value[2] = highF_ratio[np.argmax(highF_ratio)] self.eeg_ebsi_detail = [np.random.randint(1, directory.brain_ebsi_length["".join([str(self.eeg_ebsi), "_1"])]+1), np.random.randint(1, directory.brain_ebsi_length["".join([str(self.eeg_ebsi), "_2"])]+1), np.random.randint(1, directory.brain_ebsi_length["".join([str(self.eeg_ebsi), "_3"])]+1)] if int(self.real_age) < 20: # 미성년자일 경우 eeg_ebsi_detail에 어린이것만 나오도록 수정 if self.eeg_ebsi == 4: self.eeg_ebsi_detail[0] = np.random.choice([4,5]) elif self.eeg_ebsi == 5: self.eeg_ebsi_detail[0] = np.random.choice([2,4,5]) PLOT.plot_EEG_ebsi_radar([round(i, 1) for i in self.eeg_ebsi_value], self.RESULT_DIR) self.report['eeg_ebsi_value'] = [round(v,2) for v in self.eeg_ebsi_value] self.report['eeg_subtype'] = [k for k, v in directory.brain_ebsi_index.items() if (v == self.eeg_ebsi) and (k.upper() == k.lower())][0] # value to key & 한국어 가져오기 def eeg_dq(self, tmax=305): # one stop 할 때 데이터 퀄리티 먼저 전달해 주기 위한 함수 self.load_file(tmax=tmax) self.preprocess() self.cal_data_quality() return self.eeg_dq def cal_stats_eeg(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_eeg_mean = info[age]['eeg_mean'] age_eeg_std = info[age]['eeg_std'] eeg_zscore = (self.eeg_score - age_eeg_mean) / age_eeg_std percent = round(100 - st.norm.cdf(eeg_zscore) * 100) self.report['eeg_norm_percent'] = percent # temp time record definition def run_function_eeg(self, tmax=305): #start_time = time.perf_counter() # 레포트 시간 확인 self.load_file(tmax=tmax) self.preprocess() #end_time = time.perf_counter() self.cal_psd() if len(self.indices) != 0: self.cal_apf() self.zscore_apf_yn = 0 if self.peak_x < 8: self.zscore_peak_x_yn = 5 else: self.cal_cog() self.zscore_apf_yn = 10 #10 points for disadventage depending on the presence or absence of peak if self.peak_x < 8: self.zscore_peak_x_yn = 5 self.rel = self.cal_relative_power() self.faa = self.cal_faa() com = self.cal_complexity() self.cal_emoji() self.cal_ebsi_eeg() self.cal_data_quality() end_time = time.perf_counter() # print(f"EEG DQ time elapsed : {int(round((end_time - start_time) * 1000))}ms") score_list = self.cal_eeg_score() self.cal_stats_eeg() score = {'filename': self.file.split('/')[-1][:-4], 'eeg': round(self.eeg_score, 1), 'rel': round(score_list[0], 1), 'com': round(score_list[1], 1), 'faa': round(score_list[2], 1), 'apf_freq': round(score_list[3], 1), 'apf_psd': round(score_list[4], 1), 'peak ox': round(self.zscore_apf_yn, 1), 'peak <8': round(self.zscore_peak_x_yn, 1), 'eeg_dq': self.data_quality_eeg, 'eeg_emoji': str(self.eeg_emoji).split('[')[1].split(']')[0], 'eeg_ebsi': self.eeg_ebsi, 'eeg_ebsi_detail': str(self.eeg_ebsi_detail).split('[')[1].split(']')[0]} print(score) result = pd.DataFrame(score, index=[0]) result['eeg_ebsi_value'] = '' result.at[result.index[0], 'eeg_ebsi_value'] = [round(i, 1) for i in self.eeg_ebsi_value] result.to_csv(os.path.join(self.RESULT_DIR, self.file.split('/')[-1][:-4]+'_eeg.csv'), index=False, encoding='utf-8-sig') #end_time2 = time.perf_counter() #print(f"EEG code end time elapsed : {int(round((end_time - start_time) * 1000))}ms")