app.py

from fastapi import FastAPI, Request
import torch
import torch.nn as nn
import torch.optim as optim
from torchvision import transforms
from torch.utils.data import Dataset, DataLoader
from PIL import Image
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, f1_score
import segmentation_models_pytorch as smp
from torchvision.models import densenet121, DenseNet121_Weights
import requests
from io import BytesIO
from torchvision.transforms.functional import to_pil_image
import logging
import os


logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# FastAPI app instance
app = FastAPI()

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

# Load segmentation model
m1 = smp.Unet(
    encoder_name="resnet34",
    encoder_weights=None,
    in_channels=1,
    classes=1
).to(device)
m1.load_state_dict(torch.load("Segmentation_Model.pth", map_location=device))
m1.eval()

# Dataset for incremental training
class LungXrayDataset(Dataset):
    def __init__(self, entries, transform):
        self.entries = entries
        self.transform = transform
        self.label_map = {label: i for i, label in enumerate(sorted(set(s["true_label"] for s in entries)))}

    def __len__(self):
        return len(self.entries)

    def __getitem__(self, idx):
        image = self.entries[idx]["pil_image"]
        if self.transform:
            image = self.transform(image)
        label = self.label_map[self.entries[idx]["true_label"]]
        return image, label

@app.get("/")
def root():
    return {"status": "Service is up!"}

@app.post("/trigger_incremental_train")
async def trigger_train(request: Request):
    data = await request.json()
    samples = data.get("samples", [])

    if not samples or len(samples) < 100:
        return {"error": "Not enough validated samples (minimum 100 required)."}

    for entry in samples:
        try:
            logger.info(f"Downloading image: {entry['image_path']}")
            resp = requests.get(entry["image_path"], timeout=5)
            resp.raise_for_status()
            image = Image.open(BytesIO(resp.content)).convert("L").resize((256, 256))
    
            image_tensor = transforms.ToTensor()(image).unsqueeze(0).to(device)
            with torch.no_grad():
                mask = m1(image_tensor).sigmoid()
                mask = (mask > 0.5).float()
                masked = image_tensor * mask
    
            masked_rgb = masked.squeeze(0).repeat(3, 1, 1).cpu()
            masked_pil = to_pil_image(masked_rgb)
    
            entry["pil_image"] = masked_pil
    
        except Exception as e:
            error_msg = f"Failed to load image: {entry['image_path']} | Error: {e}"
            logger.error(error_msg)
            return {"error": error_msg}

    logger.info("All data loaded.")
    
    # Load models
    def load_model():
        model = densenet121(weights=None)
        model.classifier = nn.Linear(1024, 4)
        model.load_state_dict(torch.load("Classification_Model.pth", map_location=device))
        return model.to(device)

    logger.info("Loading new classification model for incremental training...")
    m2_new = load_model()
    logger.info("Loading old classification model for futher compare...")
    m2_old = load_model().eval()

    transform = transforms.Compose([
        transforms.Resize((224, 224)),
        transforms.ToTensor(),
        transforms.Normalize(mean=[0.41]*3, std=[0.16]*3)
    ])

    train_entries, val_entries = train_test_split(
        samples, test_size=0.2, stratify=[s["true_label"] for s in samples], random_state=42
    )

    train_loader = DataLoader(LungXrayDataset(train_entries, transform), batch_size=16, shuffle=True)
    val_loader = DataLoader(LungXrayDataset(val_entries, transform), batch_size=16)

    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(m2_new.parameters(), lr=1e-4)
    
    logger.info(f"Training start...")
    for epoch in range(5):
        logger.info(f"Epoch {epoch+1}/5")
        m2_new.train()
        total_loss = 0
        for imgs, labels in train_loader:
            imgs, labels = imgs.to(device), labels.to(device)
            out = m2_new(imgs)
            loss = criterion(out, labels)
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            total_loss += loss.item()
        logger.info(f"Epoch {epoch+1} complete. Loss: {round(total_loss, 4)}")

    def evaluate(model, loader):
        model.eval()
        y_true, y_pred = [], []
        with torch.no_grad():
            for imgs, labels in loader:
                imgs = imgs.to(device)
                outputs = model(imgs)
                y_pred.extend(outputs.argmax(1).cpu().numpy())
                y_true.extend(labels.numpy())
        return {
            "accuracy": round(accuracy_score(y_true, y_pred), 4),
            "f1_macro": round(f1_score(y_true, y_pred, average="macro"), 4),
        }

    logger.info(f"Evaluating old model...")
    eval_old = evaluate(m2_old, val_loader)
    logger.info(f"Evaluating new model...")
    eval_new = evaluate(m2_new, val_loader)

    os.makedirs("/tmp/data", exist_ok=True)
    model_used = "new" if eval_new["f1_macro"] > eval_old["f1_macro"] else "old"
    if model_used == "new":
        torch.save(m2_new.state_dict(), "/tmp/data/Classification_Model.pth")
        logger.info("New model saved as /tmp/data/Classification_Model.pth")

    logger.info(f"Model used: {model_used}")

    return {
        "old_model": eval_old,
        "new_model": eval_new,
        "model_used": model_used,
        "updated_model_path": "/tmp/data/Classification_Model.pth" if model_used == "new" else "unchanged"
    }