utils.py•4.03 kB
"""
VitalDB MCP Server - Utility Functions
"""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy import stats
import base64
import io
import logging
from typing import Dict, List, Tuple
import vitaldb
logger = logging.getLogger("vitaldb-utils")
# 전역 캐시
_cache = {
"loaded_cases": {},
"statistics_cache": {}
}
def create_plot_image(fig) -> str:
"""matplotlib figure를 base64 인코딩된 PNG 이미지로 변환"""
buf = io.BytesIO()
fig.savefig(buf, format='png', dpi=100, bbox_inches='tight')
buf.seek(0)
img_base64 = base64.b64encode(buf.read()).decode('utf-8')
plt.close(fig)
return img_base64
def load_case_cached(case_id: int, track_names: List[str], interval: float) -> np.ndarray:
"""캐시를 활용한 케이스 로드"""
cache_key = f"{case_id}_{','.join(sorted(track_names))}_{interval}"
if cache_key in _cache["loaded_cases"]:
logger.info(f"Using cached data for case {case_id}")
return _cache["loaded_cases"][cache_key]
logger.info(f"Loading case {case_id} from VitalDB")
vals = vitaldb.load_case(case_id, track_names, interval)
_cache["loaded_cases"][cache_key] = vals
# 캐시 크기 제한 (최대 50개 케이스)
if len(_cache["loaded_cases"]) > 50:
oldest_key = next(iter(_cache["loaded_cases"]))
del _cache["loaded_cases"][oldest_key]
return vals
def compute_statistics(data: np.ndarray) -> Dict:
"""데이터 통계 계산"""
valid_data = data[~np.isnan(data)]
if len(valid_data) == 0:
return {"error": "No valid data"}
return {
"count": len(valid_data),
"mean": float(np.mean(valid_data)),
"median": float(np.median(valid_data)),
"std": float(np.std(valid_data)),
"min": float(np.min(valid_data)),
"max": float(np.max(valid_data)),
"q25": float(np.percentile(valid_data, 25)),
"q75": float(np.percentile(valid_data, 75)),
"skewness": float(stats.skew(valid_data)),
"kurtosis": float(stats.kurtosis(valid_data))
}
def evaluate_condition(value: float, condition: str) -> bool:
"""조건 평가 (예: '>100', '>=80', '<60', '==75', '50-100')"""
try:
if '-' in condition and not condition.startswith('-'):
# 범위 조건 (예: '50-100')
parts = condition.split('-')
return float(parts[0]) <= value <= float(parts[1])
elif condition.startswith('>='):
return value >= float(condition[2:])
elif condition.startswith('<='):
return value <= float(condition[2:])
elif condition.startswith('>'):
return value > float(condition[1:])
elif condition.startswith('<'):
return value < float(condition[1:])
elif condition.startswith('=='):
return abs(value - float(condition[2:])) < 1e-6
else:
return value > float(condition)
except:
return False
def get_common_tracks():
"""일반적으로 사용되는 트랙 목록"""
return {
"생체 신호": {
"ECG_II": "심전도 Lead II",
"ECG_V5": "심전도 Lead V5",
"PLETH": "맥파",
"RESP": "호흡 파형"
},
"혈압": {
"ART": "동맥압",
"CVP": "중심정맥압",
"PAP": "폐동맥압",
"NIBP_SYS": "비침습적 수축기 혈압",
"NIBP_DIA": "비침습적 이완기 혈압",
"NIBP_MBP": "비침습적 평균 혈압"
},
"산소/가스": {
"SpO2": "산소포화도",
"EtCO2": "호기말 이산화탄소",
"FiO2": "흡입산소농도"
},
"마취/진정": {
"BIS": "뇌파지수",
"MAC": "최소폐포농도"
},
"기타": {
"HR": "심박수",
"RR": "호흡수",
"TEMP": "체온"
}
}