-
Notifications
You must be signed in to change notification settings - Fork 1
/
main.py
102 lines (80 loc) · 3.03 KB
/
main.py
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
from fastapi import FastAPI, HTTPException, status, Depends,UploadFile,File,HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from functions.genimage import generativeexecutecommand
from typing import Dict, Any
import logging
import base64
from io import BytesIO
from fastapi.responses import FileResponse
import io
import os
from dotenv import load_dotenv
load_dotenv()
logger = logging.getLogger("uvicorn")
#Load API keys
API_KEYS = os.getenv("API_KEYS")
app = FastAPI()
model_name = "Image Generation Microservice/API"
version = "v1.0.0"
## doing a simple indexing therefore no need for model(indexing "whole" documents without chunking)
class GenAgeVenBrVolInputData(BaseModel):
mriid: str = "0124"
gender: float = 0.0
age: float = 0.1
ventricular_vol: float = 0.2
brain_vol: float = 0.4
bucket_name: str = "testniis"
bucket_folder: str = "ldm"
class GenAgeVenBrVolProgress(BaseModel):
genageventbrainvol_progress: dict
@app.get("/")
def index():
return {
"message": "Endpoint to support generation of novel images given a set of population conditions."
}
@app.get("/health")
async def service_health():
"""Return service health"""
return {"status": "ok"}
@app.get("/info")
async def model_info():
"""Return model information, version, how to call"""
return {"name": model_name, "version": version}
def validate_api_key(credentials: HTTPAuthorizationCredentials = Depends(HTTPBearer(auto_error=False))):
"""Validate the API key."""
if not credentials:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Invalid or missing API Key",
)
if credentials.scheme != "Bearer" or credentials.credentials not in API_KEYS:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Invalid or missing API Key",
)
@app.post("/generategenderageventbrainvol")
async def input_index_main_text_document(
data: GenAgeVenBrVolInputData,
credentials: HTTPAuthorizationCredentials = Depends(validate_api_key)
):
"""Index whole documents...."""
config_file = "./model-zoo/models/brain_image_synthesis_latent_diffusion_model/configs/generategenderageventbrainvol_inference.json"
output_folder = "/app/MONAI/MonaiGenerativeModels/model-zoo/models/brain_image_synthesis_latent_diffusion_model/output/"
genageventbrainvol_progress = generativeexecutecommand(data.bucket_name,data.bucket_folder,output_folder,config_file,
data.gender,data.age,data.ventricular_vol,data.brain_vol,data.mriid)
output = GenAgeVenBrVolProgress(genageventbrainvol_progress=genageventbrainvol_progress) # Convert result to str
return output
origins = [
"http://localhost",
"http://localhost:5173",
"...Your Domains..."
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)