from fastapi import FastAPI, HTTPException from fastapi.responses import RedirectResponse from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel from clinical_ner import ClinicalNER import uvicorn import os # Initialize FastAPI app app = FastAPI( title="Clinical NER API", description="Named Entity Recognition and POS Tagging API using Bio_ClinicalBERT", version="1.0.0" ) # Serve static files app.mount("/app/static", StaticFiles(directory="static"), name="static") # Initialize the NER model (singleton pattern) ner_model = None @app.on_event("startup") async def startup_event(): """Load the NER model on startup""" global ner_model ner_model = ClinicalNER(use_pos=True) print("NER model loaded successfully!") # Request model class TextRequest(BaseModel): text: str class Config: json_schema_extra = { "example": { "text": "Patient presents with hypertension and diabetes. Prescribed metformin 500mg." } } # Response models class Entity(BaseModel): entity_group: str score: float word: str start: int end: int class POSToken(BaseModel): token: str lemma: str pos: str tag: str dep: str start: int end: int class BasicNERResponse(BaseModel): entities: list[Entity] count: int class PrologNERResponse(BaseModel): prolog_facts: str count: int class POSResponse(BaseModel): pos_tags: list[POSToken] count: int class PrologPOSResponse(BaseModel): prolog_facts: str count: int class CombinedResponse(BaseModel): entities: list[Entity] pos_tags: list[POSToken] entity_count: int token_count: int class PrologCombinedResponse(BaseModel): prolog_facts: str entity_count: int token_count: int @app.get("/") async def root(): return RedirectResponse(url="/browser/") @app.get("/browser/") def get_browser(): print(os.path.join("static", "browser", "index.html")) return FileResponse(os.path.join("static", "browser", "index.html")) @app.post("/ner/basic", response_model=BasicNERResponse) async def ner_basic(request: TextRequest): """ Perform basic NER on the input text. Returns a list of detected entities with their types, positions, and confidence scores. """ try: if not request.text.strip(): raise HTTPException(status_code=400, detail="Text cannot be empty") entities = ner_model.basic_ner(request.text) return { "entities": entities, "count": len(entities) } except Exception as e: raise HTTPException(status_code=500, detail=f"Error processing text: {str(e)}") @app.post("/ner/prolog", response_model=PrologNERResponse) async def ner_prolog(request: TextRequest): """ Perform NER and return results as Prolog facts. Returns Prolog facts in the format: entity(Id, Type, Word, Start, End, Score). """ try: if not request.text.strip(): raise HTTPException(status_code=400, detail="Text cannot be empty") prolog_facts = ner_model.prolog_ner(request.text) # Count the number of facts (lines) count = len(prolog_facts.split('\n')) if prolog_facts else 0 return { "prolog_facts": prolog_facts, "count": count } except Exception as e: raise HTTPException(status_code=500, detail=f"Error processing text: {str(e)}") @app.post("/pos/basic", response_model=POSResponse) async def pos_basic(request: TextRequest): """ Perform Part-of-Speech tagging on the input text. Returns a list of tokens with their POS tags, lemmas, and dependency relations. """ try: if not request.text.strip(): raise HTTPException(status_code=400, detail="Text cannot be empty") if ner_model.nlp is None: raise HTTPException(status_code=503, detail="POS tagger not available. SpaCy model not loaded.") pos_tags = ner_model.pos_tagging(request.text) return { "pos_tags": pos_tags, "count": len(pos_tags) } except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Error processing text: {str(e)}") @app.post("/pos/prolog", response_model=PrologPOSResponse) async def pos_prolog(request: TextRequest): """ Perform POS tagging and return results as Prolog facts. Returns Prolog facts in the format: pos(Id, Token, Lemma, POS, Tag, Dep, Start, End). """ try: if not request.text.strip(): raise HTTPException(status_code=400, detail="Text cannot be empty") if ner_model.nlp is None: raise HTTPException(status_code=503, detail="POS tagger not available. SpaCy model not loaded.") prolog_facts = ner_model.prolog_pos(request.text) # Count the number of facts (lines) count = len(prolog_facts.split('\n')) if prolog_facts else 0 return { "prolog_facts": prolog_facts, "count": count } except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Error processing text: {str(e)}") @app.post("/combined/basic", response_model=CombinedResponse) async def combined_basic(request: TextRequest): """ Perform both NER and POS tagging on the input text. Returns both entities and POS tags in a single response. """ try: if not request.text.strip(): raise HTTPException(status_code=400, detail="Text cannot be empty") result = ner_model.combined_analysis(request.text) return { "entities": result['entities'], "pos_tags": result['pos_tags'], "entity_count": len(result['entities']), "token_count": len(result['pos_tags']) } except Exception as e: raise HTTPException(status_code=500, detail=f"Error processing text: {str(e)}") @app.post("/combined/prolog", response_model=PrologCombinedResponse) async def combined_prolog(request: TextRequest): """ Perform both NER and POS tagging and return as Prolog facts. Returns combined Prolog facts for both entities and POS tags. """ try: if not request.text.strip(): raise HTTPException(status_code=400, detail="Text cannot be empty") prolog_facts = ner_model.prolog_combined(request.text) # Count entities and tokens lines = prolog_facts.split('\n') entity_count = len([l for l in lines if l.startswith('entity(')]) token_count = len([l for l in lines if l.startswith('pos(')]) return { "prolog_facts": prolog_facts, "entity_count": entity_count, "token_count": token_count } except Exception as e: raise HTTPException(status_code=500, detail=f"Error processing text: {str(e)}") @app.get("/health") async def health_check(): """Health check endpoint""" return { "status": "healthy", "model_loaded": ner_model is not None, "pos_available": ner_model.nlp is not None if ner_model else False } if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)