Spaces:
Sleeping
Sleeping
File size: 6,755 Bytes
6ad81c6 3480927 6ad81c6 0cc2bd2 3480927 0cc2bd2 3480927 6ad81c6 | 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 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 | import io
import os
import sys
import tempfile
from typing import List
from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.middleware.cors import CORSMiddleware
# Add the parent directory to sys.path to import bacsense_v2_package
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from bacsense_v2_package.inference import BacSense
from fastapi.responses import HTMLResponse
app = FastAPI(title="Bacsense 2.0 API")
# Setup CORS to allow requests from the React frontend
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Root landing page for Hugging Face Space visibility
@app.get("/", response_class=HTMLResponse)
async def root():
return """
<html>
<head>
<title>BacSense v2 | Precision API</title>
<style>
body { background: #020617; color: #f8fafc; font-family: -apple-system, system-ui, sans-serif; display: flex; align-items: center; justify-content: center; height: 100vh; margin: 0; }
.card { background: #0f172a; padding: 3rem; border-radius: 2rem; border: 1px solid #1e293b; text-align: center; box-shadow: 0 25px 50px -12px rgba(0,0,0,0.5); }
h1 { margin: 0 0 1rem; font-size: 2.5rem; letter-spacing: -0.025em; color: #38bdf8; }
p { color: #94a3b8; font-size: 1.125rem; margin-bottom: 2rem; }
.btn { background: #0ea5e9; color: white; padding: 0.75rem 2rem; border-radius: 0.75rem; text-decoration: none; font-weight: 600; transition: transform 0.2s; display: inline-block; }
.btn:hover { transform: scale(1.05); background: #0284c7; }
</style>
</head>
<body>
<div class="card">
<h1>🦠 BacSense v2 API</h1>
<p>The microbial classification engine is online and ready.</p>
<a href="/docs" class="btn">View API Documentation</a>
</div>
</body>
</html>
"""
# Catch-all route for SPA support (but exclude API endpoints)
@app.get("/{path_name:path}", response_class=HTMLResponse)
async def catch_all(path_name: str):
# Don't catch API endpoints - they should return 404 if not found
api_endpoints = ["docs", "redoc", "openapi.json", "health", "debug_model", "predict_batch", "api"]
if any(path_name.startswith(endpoint) for endpoint in api_endpoints):
raise HTTPException(status_code=404)
# For everything else, return the root page (useful for SPA routing)
return await root()
# Global classifier instance
CLASSIFIER = None
def get_classifier():
global CLASSIFIER
if CLASSIFIER is None:
model_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'bacsense_v2_package'))
CLASSIFIER = BacSense(model_dir=model_dir)
# Note: we skip explicit warmup here to save time; first call will be slightly slower
return CLASSIFIER
# Health check and Debugging endpoint
@app.get("/health")
async def health():
return {"status": "ok", "backend": "Hugging Face Space"}
@app.get("/debug_model")
async def debug_model():
"""Diagnostic endpoint to see why model loading might be failing."""
try:
# Check if directories exist
root = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
pkg = os.path.join(root, 'bacsense_v2_package')
exists = os.path.isdir(pkg)
# Check files
files = []
if exists:
files = os.listdir(pkg)
# Try a dummy import
try:
import tensorflow as tf
tf_version = tf.__version__
except Exception as e:
tf_version = f"Error: {e}"
# Try initializing (this might fail if deps are missing)
try:
get_classifier()
status = "Initialized OK"
error = None
except Exception as e:
status = "FAILED"
error = str(e)
import traceback
error += "\n" + traceback.format_exc()
return {
"root_dir": root,
"pkg_dir": pkg,
"pkg_exists": exists,
"pkg_files": files,
"tensorflow": tf_version,
"model_status": status,
"error_detail": error,
"sys_path": sys.path
}
except Exception as e:
return {"error": str(e)}
@app.post("/predict_batch")
async def predict_batch(files: List[UploadFile] = File(...)):
if not files or len(files) == 0:
raise HTTPException(status_code=400, detail="No files uploaded")
results = []
for file in files:
temp_path = None
try:
# Read the uploaded file into an IO stream
contents = await file.read()
# BacSense uses cv2.imread and PIL.Image.open with a file path, so we save it to disk temporarily
fd, temp_path = tempfile.mkstemp(suffix=".png")
with os.fdopen(fd, 'wb') as f:
f.write(contents)
# Process the image using lazy-loaded classifier
print(f"DEBUG: Processing file {file.filename}")
classifier = get_classifier()
result = classifier.predict(temp_path)
# Format probabilities for the frontend
confidence_pct = result["confidence"] * 100 if result["confidence"] <= 1.0 else result["confidence"]
results.append({
"filename": file.filename,
"success": True,
"prediction": result['prediction'],
"confidence": confidence_pct,
"probabilities": [
{"name": result['prediction'], "probability": confidence_pct}
],
"details": {
"gram_stain": result.get("gram", "Unknown"),
"shape": result.get("shape", "Unknown"),
"pathogenicity": result.get("risk", "Unknown")
}
})
except Exception as e:
import traceback
error_msg = f"{str(e)}\n{traceback.format_exc()}"
print(f"ERROR processing {file.filename}: {error_msg}")
results.append({
"filename": file.filename,
"success": False,
"error": str(e),
"trace": error_msg
})
finally:
# Clean up the temporary file
if temp_path and os.path.exists(temp_path):
os.remove(temp_path)
return {"results": results}
# Forced update at 02:26
|