Upload 27 files

commit to add initial codes including testing codes

backend/main.py

@@ -0,0 +1,22 @@
+from fastapi import FastAPI, UploadFile, File
+from PIL import Image
+import io
+
+from backend.predict import predict_image
+from backend.schemas import PredictionResponse
+
+app = FastAPI(
+    title="X-Ray Classification API",
+    description="Deep learning based multi-class X-ray classifier",
+    version="1.0"
+)
+# Endpoint for predicting the class of an uploaded X-ray image. 
+# Accepts an image file and returns the predicted class label and confidence score.
+
+@app.post("/predict", response_model=PredictionResponse)
+async def predict(file: UploadFile = File(...)):
+    image_bytes = await file.read()
+    image = Image.open(io.BytesIO(image_bytes))
+
+    result = predict_image(image)
+    return result

backend/model_loader.py

@@ -0,0 +1,42 @@
+import torch
+import pandas as pd
+from training.model import build_model
+from training.utils import get_device
+
+CHECKPOINT_PATH = "checkpoints/best_model.pth"
+LABEL_MAP_PATH = "data_processed/label_map.csv"
+
+# Wrapper class for loading the model and making predictions on new data instances
+class ModelWrapper:
+    def __init__(self):
+        self.device = get_device()
+
+        label_df = pd.read_csv(LABEL_MAP_PATH)
+        self.id_to_label = dict(
+            zip(label_df["label_id"], label_df["label"])
+        )
+
+        num_classes = len(self.id_to_label)
+
+        self.model = build_model(num_classes, self.device)
+        self.model.load_state_dict(
+            torch.load(CHECKPOINT_PATH, map_location=self.device)
+        )
+        self.model.eval()
+
+    def predict(self, image_tensor):
+        with torch.no_grad():
+            image_tensor = image_tensor.to(self.device)
+            outputs = self.model(image_tensor)
+            probs = torch.softmax(outputs, dim=1)
+
+            confidence, pred_id = torch.max(probs, dim=1)
+
+        return {
+            "label_id": int(pred_id.item()),
+            "label_name": self.id_to_label[int(pred_id.item())],
+            "confidence": float(confidence.item())
+        }
+
+
+model_wrapper = ModelWrapper()

backend/predict.py

@@ -0,0 +1,19 @@
+from PIL import Image
+import torch
+from torchvision import transforms
+from backend.model_loader import model_wrapper
+
+# Defining the image transformation pipeline for preprocessing
+transform = transforms.Compose([
+    transforms.Resize((224, 224)),
+    transforms.ToTensor(),
+    transforms.Normalize(
+        mean=[0.485, 0.456, 0.406],
+        std=[0.229, 0.224, 0.225]
+    )
+])
+# Function to predict the class of an input image using the loaded best model
+def predict_image(image: Image.Image):
+    image = image.convert("RGB")
+    tensor = transform(image).unsqueeze(0)
+    return model_wrapper.predict(tensor)

backend/schemas.py

@@ -0,0 +1,8 @@
+#Module that defines the schema for prediction responses on the API.
+from pydantic import BaseModel
+
+
+class PredictionResponse(BaseModel):
+    label_id: int
+    label_name: str
+    confidence: float

checkpoints/best_model.pth

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:d83d0507e2b50d687569e7ccab21287169f4a13ba3cd7441437a2001cefa82b6
+size 94842430

requirements.txt

@@ -0,0 +1,9 @@
+gradio
+fastapi
+torch==2.1.0
+torchvision==0.16.0
+pillow
+numpy
+pandas
+scikit-learn
+python-multipart

scripts/01_merge_datasets.py

@@ -0,0 +1,56 @@
+import os
+import shutil
+import csv
+from pathlib import Path
+
+RAW_ROOT = Path("raw_data")
+OUT_IMG = Path("data_merged/images")
+OUT_CSV = Path("data_merged/metadata_raw.csv")
+
+OUT_IMG.mkdir(parents=True, exist_ok=True)
+
+rows = []
+img_id = 0
+VALID_EXT = (".png", ".jpg", ".jpeg")
+
+def merge_any_dataset(dataset_name, base_path):
+    global img_id
+    for root, _, files in os.walk(base_path):
+        for f in files:
+            if not f.lower().endswith(VALID_EXT):
+                continue
+
+            src = Path(root) / f
+            class_name = Path(root).name
+
+            new_name = f"{dataset_name}__{class_name}__{img_id}{src.suffix}"
+            dst = OUT_IMG / new_name
+
+            shutil.copy(src, dst)
+
+            rows.append({
+                "image_id": img_id,
+                "filename": new_name,
+                "label": class_name,
+                "source": dataset_name
+            })
+
+            img_id += 1
+
+# Merging all datasets found in RAW_ROOT
+for item in RAW_ROOT.iterdir():
+    if item.is_dir():
+        merge_any_dataset(item.name, item)
+
+# writing out the CSV
+OUT_CSV.parent.mkdir(parents=True, exist_ok=True)
+with open(OUT_CSV, "w", newline="", encoding="utf-8") as f:
+    writer = csv.DictWriter(
+        f,
+        fieldnames=["image_id", "filename", "label", "source"]
+    )
+    writer.writeheader()
+    writer.writerows(rows)
+
+print("Merged dataset created")
+print("Images:", len(rows))

scripts/02_resize_images.py

@@ -0,0 +1,30 @@
+import cv2
+import pandas as pd
+from pathlib import Path
+
+IN_IMG = Path("data_merged/images")
+OUT_IMG = Path("data_processed/images")
+IN_CSV = Path("data_merged/metadata_raw.csv")
+OUT_CSV = Path("data_processed/metadata_resized.csv")
+
+OUT_IMG.mkdir(parents=True, exist_ok=True)
+
+df = pd.read_csv(IN_CSV)
+kept_rows = []
+
+for _, row in df.iterrows():
+    src = IN_IMG / row["filename"]
+    dst = OUT_IMG / row["filename"]
+
+    img = cv2.imread(str(src))
+    if img is None:
+        continue
+
+    img = cv2.resize(img, (224, 224))
+    cv2.imwrite(str(dst), img)
+
+    kept_rows.append(row)
+
+pd.DataFrame(kept_rows).to_csv(OUT_CSV, index=False)
+
+print("Images kept:", len(kept_rows))

scripts/03_create_metadata.py

@@ -0,0 +1,22 @@
+import pandas as pd
+from pathlib import Path
+
+IN_CSV = Path("data_processed/metadata_resized.csv")
+OUT_CSV = Path("data_processed/metadata_final.csv")
+MAP_CSV = Path("data_processed/label_map.csv")
+
+df = pd.read_csv(IN_CSV)
+
+labels = sorted(df["label"].unique())
+label_to_id = {label: i for i, label in enumerate(labels)}
+
+df["label_id"] = df["label"].map(label_to_id)
+
+df.to_csv(OUT_CSV, index=False)
+
+pd.DataFrame({
+    "label": labels,
+    "label_id": [label_to_id[l] for l in labels]
+}).to_csv(MAP_CSV, index=False)
+
+print("Total classes:", len(labels))

training/dataloader.py

@@ -0,0 +1,47 @@
+#Module for creating data loaders for training and validation datasets.
+from torch.utils.data import DataLoader, random_split
+from dataset import XRayDataset
+
+def get_dataloaders(
+    csv_path,
+    images_dir,
+    batch_size=32,
+    val_split=0.2
+):
+    full_dataset = XRayDataset(
+        csv_path=csv_path,
+        images_dir=images_dir,
+        train=True
+    )
+
+    val_size = int(len(full_dataset) * val_split)
+    train_size = len(full_dataset) - val_size
+
+    train_ds, val_ds = random_split(
+        full_dataset,
+        [train_size, val_size]
+    )
+
+    # Disable augmentation for validation dataset so that we only apply normalization
+    val_ds.dataset.transform = XRayDataset(
+        csv_path,
+        images_dir,
+        train=False
+    ).transform
+
+    train_loader = DataLoader(
+    train_ds,
+    batch_size=batch_size,
+    shuffle=True,
+    num_workers=0
+)
+
+    val_loader = DataLoader(
+    val_ds,
+    batch_size=batch_size,
+    shuffle=False,
+    num_workers=0
+)
+
+
+    return train_loader, val_loader

training/dataset.py

@@ -0,0 +1,48 @@
+from torch.utils.data import Dataset
+from torchvision import transforms
+from PIL import Image
+import pandas as pd
+from pathlib import Path
+
+# Custom Dataset for X-Ray Images with Augmentations for Training and Standard Transformations for Validation
+
+class XRayDataset(Dataset):
+    def __init__(self, csv_path, images_dir, train=True):
+        self.df = pd.read_csv(csv_path)
+        self.images_dir = Path(images_dir)
+
+        if train:
+            self.transform = transforms.Compose([
+                transforms.Resize((224, 224)),
+                transforms.RandomHorizontalFlip(p=0.5),
+                transforms.RandomRotation(15),
+                transforms.RandomResizedCrop(224, scale=(0.85, 1.0)),
+                transforms.ColorJitter(brightness=0.1, contrast=0.1),
+                transforms.ToTensor(),
+                transforms.Normalize(
+                    mean=[0.485, 0.456, 0.406],
+                    std=[0.229, 0.224, 0.225]
+                )
+            ])
+        else:
+            self.transform = transforms.Compose([
+                transforms.Resize((224, 224)),
+                transforms.ToTensor(),
+                transforms.Normalize(
+                    mean=[0.485, 0.456, 0.406],
+                    std=[0.229, 0.224, 0.225]
+                )
+            ])
+
+    def __len__(self):
+        return len(self.df)
+
+    def __getitem__(self, idx):
+        row = self.df.iloc[idx]
+        img_path = self.images_dir / row["filename"]
+        label = row["label_id"]
+
+        image = Image.open(img_path).convert("RGB")
+        image = self.transform(image)
+
+        return image, label

training/evaluate.py

@@ -0,0 +1,51 @@
+import torch
+import pandas as pd
+import seaborn as sns
+import matplotlib.pyplot as plt
+from sklearn.metrics import confusion_matrix, classification_report
+
+from model import build_model
+from dataloader import get_dataloaders
+from utils import get_device
+
+CSV_PATH = "data_processed/metadata_final.csv"
+IMG_DIR = "data_processed/images"
+CHECKPOINT_PATH = "checkpoints/best_model.pth"
+
+device = get_device()
+
+df = pd.read_csv(CSV_PATH)
+num_classes = df["label_id"].nunique()
+
+model = build_model(num_classes, device)
+model.load_state_dict(torch.load(CHECKPOINT_PATH))
+model.eval()
+
+_, val_loader = get_dataloaders(
+    csv_path=CSV_PATH,
+    images_dir=IMG_DIR,
+    batch_size=32
+)
+
+y_true, y_pred = [], []
+
+with torch.no_grad():
+    for images, labels in val_loader:
+        images = images.to(device)
+        outputs = model(images)
+        preds = outputs.argmax(dim=1).cpu().numpy()
+
+        y_pred.extend(preds)
+        y_true.extend(labels.numpy())
+
+cm = confusion_matrix(y_true, y_pred)
+
+plt.figure(figsize=(14, 12))
+sns.heatmap(cm, cmap="Blues", xticklabels=False, yticklabels=False)
+plt.title("Confusion Matrix")
+plt.xlabel("Predicted")
+plt.ylabel("True")
+plt.show()
+
+print("\nClassification Report:")
+print(classification_report(y_true, y_pred))

training/model.py

@@ -0,0 +1,23 @@
+import torch
+import torch.nn as nn
+from torchvision import models
+
+def build_model(num_classes: int, device: torch.device):
+    model = models.resnet50(weights=models.ResNet50_Weights.IMAGENET1K_V2)
+
+    # Freeze early layers
+    for name, param in model.named_parameters():
+        if not (
+            name.startswith("layer3") or
+            name.startswith("layer4") or
+            name.startswith("fc")
+        ):
+            param.requires_grad = False
+
+
+
+    # Replace classifier
+    in_features = model.fc.in_features
+    model.fc = nn.Linear(in_features, num_classes)
+
+    return model.to(device)

training/test_dataset.py

@@ -0,0 +1,13 @@
+#Testing the XRayDataset class for correctness and functionality
+from dataset import XRayDataset
+
+ds = XRayDataset(
+    csv_path="data_processed/metadata_final.csv",
+    images_dir="data_processed/images",
+    train=True
+)
+
+print("Total samples:", len(ds))
+img, label = ds[0]
+print("Image shape:", img.shape)
+print("Label:", label)

training/test_loader.py

@@ -0,0 +1,12 @@
+#Testing the dataloader functionality
+from dataloader import get_dataloaders
+
+train_loader, val_loader = get_dataloaders(
+    csv_path="data_processed/metadata_final.csv",
+    images_dir="data_processed/images",
+    batch_size=32
+)
+
+images, labels = next(iter(train_loader))
+print("Batch image shape:", images.shape)
+print("Batch labels shape:", labels.shape)

training/test_model.py

@@ -0,0 +1,26 @@
+import torch.nn as nn
+from torchvision import models
+
+
+def build_model(num_classes, device):
+    model = models.resnet18(
+        weights=models.ResNet18_Weights.IMAGENET1K_V1
+    )
+
+    # Freezing everything
+    for param in model.parameters():
+        param.requires_grad = False
+
+    # Unfreezing deeper layers
+    for param in model.layer3.parameters():
+        param.requires_grad = True
+
+    for param in model.layer4.parameters():
+        param.requires_grad = True
+
+    # Replacing classifier for our number of classes
+    in_features = model.fc.in_features
+    model.fc = nn.Linear(in_features, num_classes)
+
+    model = model.to(device)
+    return model

training/train.py

@@ -0,0 +1,160 @@
+import os
+import torch
+import torch.nn as nn
+import torch.optim as optim
+import pandas as pd
+import numpy as np
+from tqdm import tqdm
+
+from dataloader import get_dataloaders
+from model import build_model
+from utils import get_device, accuracy
+
+
+def compute_class_weights(csv_path):
+    df = pd.read_csv(csv_path)
+
+    class_counts = df["label_id"].value_counts().sort_index()
+    total_samples = class_counts.sum()
+
+    class_counts = torch.tensor(class_counts.values, dtype=torch.float32)
+
+    # Soft inverse-frequency weighting
+    weights = total_samples / class_counts
+
+    # Log-scale to reduce extremes
+    weights = torch.log1p(weights)
+
+    # Normalize
+    weights = weights / weights.mean()
+
+    # 🔒 Cap extreme weights (critical)
+    weights = torch.clamp(weights, max=3.0)
+
+    return weights
+
+
+
+# Train and validation functions for one epoch each
+
+def train_one_epoch(model, loader, criterion, optimizer, device):
+    model.train()
+    total_loss, total_acc = 0.0, 0.0
+
+    for images, labels in tqdm(loader, desc="Training", leave=False):
+        images, labels = images.to(device), labels.to(device)
+
+        optimizer.zero_grad()
+        outputs = model(images)
+        loss = criterion(outputs, labels)
+
+        loss.backward()
+        optimizer.step()
+
+        total_loss += loss.item()
+        total_acc += accuracy(outputs, labels)
+
+    return total_loss / len(loader), total_acc / len(loader)
+
+
+def validate_one_epoch(model, loader, criterion, device):
+    model.eval()
+    total_loss, total_acc = 0.0, 0.0
+
+    with torch.no_grad():
+        for images, labels in tqdm(loader, desc="Validation", leave=False):
+            images, labels = images.to(device), labels.to(device)
+            outputs = model(images)
+            loss = criterion(outputs, labels)
+
+            total_loss += loss.item()
+            total_acc += accuracy(outputs, labels)
+
+    return total_loss / len(loader), total_acc / len(loader)
+
+
+def main():
+    #Hyperparameters and paths
+    BATCH_SIZE = 32
+    EPOCHS = 20
+    LR = 1e-4
+    PATIENCE = 4
+
+    CSV_PATH = "data_processed/metadata_final.csv"
+    IMG_DIR = "data_processed/images"
+    CHECKPOINT_DIR = "checkpoints"
+    CHECKPOINT_PATH = f"{CHECKPOINT_DIR}/best_model.pth"
+
+    os.makedirs(CHECKPOINT_DIR, exist_ok=True)
+
+    #Setup
+    device = get_device()
+    print("Using device:", device)
+
+    df = pd.read_csv(CSV_PATH)
+    num_classes = df["label_id"].nunique()
+
+    train_loader, val_loader = get_dataloaders(
+        csv_path=CSV_PATH,
+        images_dir=IMG_DIR,
+        batch_size=BATCH_SIZE
+    )
+
+    model = build_model(num_classes, device)
+
+    class_weights = compute_class_weights(CSV_PATH).to(device)
+    criterion = nn.CrossEntropyLoss(
+        weight=class_weights,
+        label_smoothing=0.02
+    )
+
+
+    optimizer = torch.optim.AdamW(
+        model.parameters(),
+        lr=LR,
+        weight_decay=1e-4
+    )
+
+    # Learning rate scheduler so that lr reduces if val loss plateaus
+    scheduler = optim.lr_scheduler.ReduceLROnPlateau(
+        optimizer, mode="min", patience=2, factor=0.5
+    )
+
+    best_val_loss = float("inf")
+    epochs_without_improvement = 0
+
+    # Training loop with early stopping to prevent overfitting
+    for epoch in range(EPOCHS):
+        print(f"\nEpoch [{epoch + 1}/{EPOCHS}]")
+
+        train_loss, train_acc = train_one_epoch(
+            model, train_loader, criterion, optimizer, device
+        )
+
+        val_loss, val_acc = validate_one_epoch(
+            model, val_loader, criterion, device
+        )
+
+        scheduler.step(val_loss)
+
+        print(
+            f"Train Loss: {train_loss:.4f} | Train Acc: {train_acc:.4f} | "
+            f"Val Loss: {val_loss:.4f} | Val Acc: {val_acc:.4f}"
+        )
+
+        if val_loss < best_val_loss:
+            best_val_loss = val_loss
+            epochs_without_improvement = 0
+            torch.save(model.state_dict(), CHECKPOINT_PATH)
+            print("Best model saved")
+        else:
+            epochs_without_improvement += 1
+            if epochs_without_improvement >= PATIENCE:
+                print("Early stopping triggered")
+                break
+
+    print("\nTraining is complete.")
+
+
+if __name__ == "__main__":
+    main()

training/utils.py

@@ -0,0 +1,13 @@
+# Utility functions for training machine learning models using PyTorch and calculating accuracy.
+import torch
+
+def get_device():
+    if torch.cuda.is_available():
+        return torch.device("cuda")
+    return torch.device("cpu")
+
+
+def accuracy(outputs, labels):
+    preds = outputs.argmax(dim=1)
+    correct = (preds == labels).sum().item()
+    return correct / labels.size(0)