xrayvision-backend / app /services /chest_model.py
zohaibcodez's picture
initial deploy
ce4fddb
Raw
History Blame Contribute Delete
4.36 kB
"""Chest pathology detection using TorchXRayVision DenseNet121.
This model is pretrained on 700K+ clinical X-ray images and outputs
probabilities for 18 different chest pathologies.
"""
from __future__ import annotations
import numpy as np
import torch
import logging
logger = logging.getLogger(__name__)
# Lazy-loaded model singleton
_model = None
_pathology_names: list[str] = []
def _get_model():
"""Load the DenseNet121 model (lazy singleton)."""
global _model, _pathology_names
if _model is None:
logger.info("Loading TorchXRayVision DenseNet121 model...")
try:
import torchxrayvision as xrv
_model = xrv.models.DenseNet(weights="densenet121-res224-all")
_model.eval()
_pathology_names = list(_model.pathologies)
logger.info(f"DenseNet121 loaded. Pathologies: {_pathology_names}")
except Exception as e:
logger.error(f"Failed to load DenseNet121: {e}")
raise
return _model
def predict_chest_pathologies(preprocessed_image: np.ndarray,
confidence_threshold: float = 0.40) -> list[dict]:
"""Run chest pathology inference.
Args:
preprocessed_image: numpy array of shape (1, 1, 224, 224)
normalized to [-1024, 1024] range.
confidence_threshold: minimum probability to include in results.
Returns:
List of findings, each with name, confidence, severity, model, region.
"""
model = _get_model()
with torch.no_grad():
tensor_input = torch.from_numpy(preprocessed_image).float()
outputs = model(tensor_input)
probabilities = outputs.cpu().numpy()[0]
findings = []
for i, (name, prob) in enumerate(zip(_pathology_names, probabilities)):
confidence = float(prob) * 100 # Convert to percentage
if confidence < confidence_threshold * 100:
continue
# Determine severity based on confidence
if confidence >= 80:
severity = "high"
color = "destructive"
elif confidence >= 60:
severity = "moderate"
color = "warning"
else:
severity = "low"
color = "info"
# Map pathology to anatomical region
region = _get_region(name)
icd_code = _get_icd_code(name)
findings.append({
"name": name,
"confidence": round(confidence, 1),
"severity": severity,
"model": "DenseNet121",
"region": region,
"icd_code": icd_code,
"color": color,
})
# Sort by confidence descending
findings.sort(key=lambda x: x["confidence"], reverse=True)
return findings
def _get_region(pathology: str) -> str:
"""Map pathology name to anatomical region."""
region_map = {
"Atelectasis": "Lung parenchyma",
"Cardiomegaly": "Mediastinum",
"Consolidation": "Lung parenchyma",
"Edema": "Bilateral",
"Effusion": "Pleural space",
"Emphysema": "Lung parenchyma",
"Fibrosis": "Lung parenchyma",
"Hernia": "Diaphragm",
"Infiltration": "Lung parenchyma",
"Mass": "Lung parenchyma",
"Nodule": "Lung parenchyma",
"Pleural_Thickening": "Pleural space",
"Pneumonia": "Lung parenchyma",
"Pneumothorax": "Pleural space",
"Enlarged Cardiomediastinum": "Mediastinum",
"Lung Opacity": "Lung parenchyma",
"Lung Lesion": "Lung parenchyma",
"Fracture": "Rib cage",
}
return region_map.get(pathology, "Chest")
def _get_icd_code(pathology: str) -> str:
"""Map pathology name to ICD-10 code."""
icd_map = {
"Atelectasis": "J98.1",
"Cardiomegaly": "I51.7",
"Consolidation": "J18.9",
"Edema": "J81",
"Effusion": "J90",
"Emphysema": "J43.9",
"Fibrosis": "J84.1",
"Hernia": "K44.9",
"Infiltration": "R91.8",
"Mass": "R91.1",
"Nodule": "R91.1",
"Pleural_Thickening": "J92.9",
"Pneumonia": "J18.9",
"Pneumothorax": "J93.9",
"Enlarged Cardiomediastinum": "R93.1",
"Lung Opacity": "R91.8",
"Lung Lesion": "R91.1",
"Fracture": "S22.9",
}
return icd_map.get(pathology, "R93.1")