Spaces:
Paused
Paused
File size: 3,426 Bytes
d5f2660 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 |
"""Medical AI models client"""
import aiohttp
from typing import Optional, Dict, Any
from .logger import setup_logger
from .config import MODELS_SERVER_URL, MODELS_SERVER_TIMEOUT
logger = setup_logger(__name__)
async def classify_image(image_path: str, model: str) -> Optional[Dict[str, Any]]:
"""Classify medical image"""
try:
logger.info(f"🔬 Classification | Model: {model} | Image: {image_path.split('/')[-1]}")
timeout = aiohttp.ClientTimeout(total=MODELS_SERVER_TIMEOUT)
async with aiohttp.ClientSession(timeout=timeout) as session:
payload = {"image_path": image_path, "model": model}
async with session.post(f"{MODELS_SERVER_URL}/classify", json=payload) as resp:
if resp.status == 200:
result = await resp.json()
top_pred = result['top_prediction']
logger.info(f"✅ Classification | {model} → {top_pred['class_name']} ({top_pred['confidence']*100:.1f}%)")
return result
else:
error = await resp.text()
logger.error(f"❌ Classification failed: {resp.status} - {error}")
return None
except Exception as e:
logger.error(f"❌ Classification error: {e}")
return None
async def detect_objects(image_path: str, model: str) -> Optional[Dict[str, Any]]:
"""Detect objects in medical image"""
try:
logger.info(f"🔍 Detection | Model: {model} | Image: {image_path.split('/')[-1]}")
timeout = aiohttp.ClientTimeout(total=MODELS_SERVER_TIMEOUT)
async with aiohttp.ClientSession(timeout=timeout) as session:
payload = {"image_path": image_path, "model": model}
async with session.post(f"{MODELS_SERVER_URL}/detect", json=payload) as resp:
if resp.status == 200:
result = await resp.json()
logger.info(f"✅ Detection | {model} → {result['total_detections']} objects found")
return result
else:
error = await resp.text()
logger.error(f"❌ Detection failed: {resp.status} - {error}")
return None
except Exception as e:
logger.error(f"❌ Detection error: {e}")
return None
async def segment_case(case_path: str, model: str = "brats") -> Optional[Dict[str, Any]]:
"""Segment 3D brain tumor case"""
try:
case_name = case_path.split('/')[-1]
logger.info(f"📊 Segmentation | Model: {model} | Case: {case_name}")
timeout = aiohttp.ClientTimeout(total=MODELS_SERVER_TIMEOUT)
async with aiohttp.ClientSession(timeout=timeout) as session:
payload = {"case_path": case_path, "model": model}
async with session.post(f"{MODELS_SERVER_URL}/segment", json=payload) as resp:
if resp.status == 200:
result = await resp.json()
logger.info(f"✅ Segmentation | {model} → Avg Dice: {result['average_dice']:.3f}")
return result
else:
error = await resp.text()
logger.error(f"❌ Segmentation failed: {resp.status} - {error}")
return None
except Exception as e:
logger.error(f"❌ Segmentation error: {e}")
return None
|