Spaces:
Running
Running
File size: 3,277 Bytes
79b0bef | 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 | """
src/api/routes/icd.py
────────────────────────────────────────────────────────────────
Endpoints for ICD-10 code mapping.
Routes
──────
POST /icd/lookup — map a free-text entity to ICD-10 codes
GET /icd/top — most frequently assigned codes in the dataset
────────────────────────────────────────────────────────────────
"""
from __future__ import annotations
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from src.api.schemas import ICD10LookupRequest, ICD10LookupResponse, ICD10MatchResponse
from src.db.connection import get_db_session
from src.db.repository import ICD10Repository
from src.utils.logger import get_logger
logger = get_logger(__name__)
router = APIRouter(prefix="/icd", tags=["ICD-10"])
# Module-level singleton — loaded once, reused per request
_mapper = None
def _get_mapper():
"""Return the ICD-10 mapper, loading it if needed."""
global _mapper
if _mapper is None:
from src.nlp.icd_mapper import ICD10Mapper
try:
_mapper = ICD10Mapper()
except FileNotFoundError as exc:
logger.warning("ICD-10 mapper unavailable: %s", exc)
return _mapper
@router.post(
"/lookup",
response_model = ICD10LookupResponse,
summary = "Map clinical text to ICD-10 codes",
)
def lookup_icd10(payload: ICD10LookupRequest) -> ICD10LookupResponse:
"""Map a free-text clinical entity to ICD-10-CM codes.
Tries exact → fuzzy → embedding matching in order.
Returns the top-k candidates with confidence scores.
Args:
payload: :class:`ICD10LookupRequest` with the entity text.
Returns:
:class:`ICD10LookupResponse` with ranked ICD-10 candidates.
"""
mapper = _get_mapper()
if mapper is None:
return ICD10LookupResponse(query=payload.text, matches=[])
# Override top_k from the request
mapper._top_k = payload.top_k
matches = mapper.map(payload.text)
return ICD10LookupResponse(
query = payload.text,
matches = [ICD10MatchResponse(**m.to_dict()) for m in matches],
)
@router.get(
"/top",
response_model = list[ICD10MatchResponse],
summary = "Most frequently assigned ICD-10 codes",
)
def top_icd10_codes(
limit: int = 20,
db: Session = Depends(get_db_session),
) -> list[ICD10MatchResponse]:
"""Return the most frequently assigned ICD-10 codes in the dataset.
Used by the dashboard to show which diagnoses dominate the corpus.
Args:
limit: Number of top codes to return (default 20).
Returns:
List of :class:`ICD10MatchResponse` objects ranked by frequency.
"""
rows = ICD10Repository.top_codes(db, limit=limit)
return [
ICD10MatchResponse(
icd10_code = code,
description = desc,
confidence = 1.0,
match_method = "lookup",
rank = i + 1,
)
for i, (code, desc, _) in enumerate(rows)
]
|