utils.py•4.92 kB
"""
VitalDB MCP Server - Utility Functions
"""
import base64
import io
import logging
from typing import Any, Dict, List, Tuple
logger = logging.getLogger("vitaldb-utils")
# 전역 캐시
_cache = {
"loaded_cases": {},
"statistics_cache": {}
}
# 공통 트랙 캐시 (지연 구축)
_common_tracks_cache = None
def create_plot_image(fig) -> str:
"""matplotlib figure를 base64 인코딩된 PNG 이미지로 변환"""
# 지연 임포트: 실제 호출 시점에만 matplotlib 사용
import matplotlib.pyplot as plt
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) -> Any:
"""캐시를 활용한 케이스 로드"""
# 지연 임포트: 호출 시점에만 vitaldb 접근
import importlib
vitaldb = importlib.import_module("vitaldb")
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: Any) -> Dict:
"""데이터 통계 계산"""
import numpy as np
from scipy import stats
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():
"""일반적으로 사용되는 트랙 목록을 지연 생성하고 캐시합니다.
무거운 I/O가 필요하다면 이 함수 내부에서 최초 1회만 수행하고
결과를 캐싱해 이후 호출에서는 메모리 캐시를 반환합니다.
현재 구현은 정적인 딕셔너리를 캐시합니다.
"""
global _common_tracks_cache
if _common_tracks_cache is None:
# 필요 시 여기에서만 지연 임포트/네트워크 접근을 수행하세요.
# 예) from vitaldb import something
_common_tracks_cache = {
"생체 신호": {
"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": "체온",
},
}
return _common_tracks_cache