Spaces:
Runtime error
Runtime error
File size: 4,374 Bytes
87e1988 | 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 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | import requests
import io
from PIL import Image
from typing import Dict, Any
import logging
logger = logging.getLogger(__name__)
class DermaScanAPIClient:
"""Client for external DermaScan AI API"""
def __init__(self, base_url: str = "https://omarelrayes-api.hf.space"):
self.base_url = base_url.rstrip('/')
self.timeout = 60
def health_check(self) -> bool:
"""Check if external API is healthy"""
try:
response = requests.get(f"{self.base_url}/health", timeout=5)
return response.status_code == 200
except Exception as e:
logger.error(f"Health check failed: {e}")
return False
def upload_and_analyze_image(self, image: Image.Image, thread_id: str) -> Dict[str, Any]:
"""
Upload image and get full analysis (classification + segmentation)
Returns dict with: label, confidence_pct, infection_pct, images
"""
try:
# Convert PIL Image to bytes
img_byte_arr = io.BytesIO()
image.save(img_byte_arr, format='PNG')
img_byte_arr.seek(0)
# Upload and analyze
files = {'file': ('image.png', img_byte_arr, 'image/png')}
data = {'thread_id': thread_id}
response = requests.post(
f"{self.base_url}/analyze-image",
files=files,
data=data,
timeout=self.timeout
)
response.raise_for_status()
result = response.json()
if not result.get('ok'):
return {'error': result.get('error', 'Analysis failed')}
return {
'label': result.get('label', 'Unknown'),
'confidence_pct': result.get('confidence_pct', 0),
'infection_pct': result.get('infection_pct', 0),
'images': result.get('images', {}),
'patient': result.get('patient', {}),
}
except requests.exceptions.Timeout:
return {'error': 'API request timed out'}
except requests.exceptions.RequestException as e:
return {'error': f'API request failed: {str(e)}'}
except Exception as e:
logger.error(f"Unexpected error in upload_and_analyze_image: {e}", exc_info=True)
return {'error': f'Unexpected error: {str(e)}'}
def book_appointment(self, thread_id: str, city: str = "") -> Dict[str, Any]:
"""Book appointment via external API"""
try:
response = requests.post(
f"{self.base_url}/book-appointment",
json={'thread_id': thread_id, 'city': city},
timeout=30
)
response.raise_for_status()
return response.json()
except Exception as e:
return {'ok': False, 'error': str(e)}
def get_patient_image(self, thread_id: str) -> Dict[str, Any]:
"""Get patient's uploaded image"""
try:
response = requests.get(
f"{self.base_url}/patient-image",
params={'thread_id': thread_id},
timeout=30
)
response.raise_for_status()
return response.json()
except Exception as e:
return {'ok': False, 'error': str(e)}
def segment_patient_image(self, thread_id: str) -> Dict[str, Any]:
"""Segment patient's image"""
try:
response = requests.post(
f"{self.base_url}/segment-patient-image",
json={'thread_id': thread_id},
timeout=60
)
response.raise_for_status()
return response.json()
except Exception as e:
return {'ok': False, 'error': str(e)}
def reanalyze_patient_image(self, thread_id: str) -> Dict[str, Any]:
"""Reanalyze patient's image"""
try:
response = requests.post(
f"{self.base_url}/reanalyze-patient-image",
json={'thread_id': thread_id},
timeout=60
)
response.raise_for_status()
return response.json()
except Exception as e:
return {'ok': False, 'error': str(e)} |