import pathlib as pl import time from threading import Event from typing import Callable, Tuple, cast import numpy as np import pandas as pd import torch import torch.nn as nn from torch.utils.data import DataLoader from model.dataset import ADNIDataset from util.progress import ProgressTracker from util.ui_logger import PipelineLogger type TrainMetrics = Tuple[ float, float, float, float ] # (train_loss, val_loss, train_acc, val_acc) type TestMetrics = Tuple[float, float] # (test_loss, test_acc) type KLLossFn = Callable[[nn.Module], torch.Tensor | None] def _model_device(model: nn.Module) -> torch.device: first_param = next(model.parameters(), None) if first_param is None: return torch.device("cpu") return first_param.device def _move_batch_to_model_device( model: nn.Module, mri: torch.Tensor, xls: torch.Tensor, targets: torch.Tensor, ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]: device = _model_device(model) mri = mri.to(device, non_blocking=True) xls = xls.to(device, non_blocking=True) targets = targets.to(device, non_blocking=True) return mri, xls, targets def _batch_correct_and_total( outputs: torch.Tensor, targets: torch.Tensor, ) -> Tuple[int, int]: if outputs.ndim > 1 and outputs.size(-1) > 1: predicted_classes = outputs.argmax(dim=1) if targets.ndim > 1 and targets.size(-1) > 1: target_classes = targets.argmax(dim=1) else: target_classes = targets.long().view(-1) correct = int((predicted_classes == target_classes).sum().item()) total = int(target_classes.numel()) return correct, total predicted = (outputs > 0.5).to(dtype=targets.dtype) correct = int((predicted == targets).sum().item()) total = int(targets.numel()) return correct, total def _check_control_events( stop_event: Event | None, pause_event: Event | None, ) -> None: if stop_event is not None and stop_event.is_set(): raise InterruptedError("Pipeline execution stopped by user.") while pause_event is not None and pause_event.is_set(): time.sleep(0.5) if stop_event is not None and stop_event.is_set(): raise InterruptedError("Pipeline execution stopped by user while paused.") def test_model( model: nn.Module, test_loader: DataLoader[ADNIDataset], criterion: nn.Module, progress: ProgressTracker, log: PipelineLogger, stop_event: Event | None = None, pause_event: Event | None = None, ) -> TestMetrics: """ Tests the model on the test dataset. Args: model (nn.Module): The model to test. test_loader (DataLoader[ADNIDataset]): DataLoader for the test dataset. criterion (nn.Module): Loss function to compute the loss. Returns: TrainMetrics: A tuple containing the test loss and test accuracy. """ model.eval() test_loss = 0.0 correct = 0 total = 0 total_samples = 0 test_progress = progress.get_sub_tracker("Testing Batches") test_progress.update(total=len(test_loader), advance=0) with torch.no_grad(): for _, (mri, xls, targets, _) in enumerate(test_loader): _check_control_events(stop_event=stop_event, pause_event=pause_event) mri, xls, targets = _move_batch_to_model_device(model, mri, xls, targets) outputs = model((mri, xls)) loss = criterion(outputs, targets) batch_size = mri.size(0) test_loss += loss.item() * batch_size total_samples += batch_size # Calculate accuracy batch_correct, batch_total = _batch_correct_and_total(outputs, targets) correct += batch_correct total += batch_total test_progress.update(total=len(test_loader), advance=1) test_loss = test_loss / total_samples if total_samples > 0 else 0.0 test_acc = correct / total if total > 0 else 0.0 return test_loss, test_acc def train_epoch( model: nn.Module, train_loader: DataLoader[ADNIDataset], val_loader: DataLoader[ADNIDataset], optimizer: torch.optim.Optimizer, criterion: nn.Module, log: PipelineLogger, progress: ProgressTracker, stop_event: Event | None = None, pause_event: Event | None = None, ) -> Tuple[float, float, float, float]: """ Trains the model for one epoch and evaluates it on the validation set. Args: model (nn.Module): The model to train. train_loader (DataLoader[ADNIDataset]): DataLoader for the training dataset. val_loader (DataLoader[ADNIDataset]): DataLoader for the validation dataset. optimizer (torch.optim.Optimizer): Optimizer for updating model parameters. criterion (nn.Module): Loss function to compute the loss. Returns: Tuple[float, float, float, float]: A tuple containing the training loss, validation loss, training accuracy, and validation accuracy. """ model.train() train_loss = 0.0 train_correct = 0 train_total = 0 train_total_samples = 0 # Training loop - with progress bar for the training batches train_progress = progress.get_sub_tracker("Training Batches") train_progress.reset() train_progress.set_title("Training Batches") train_progress.update(total=len(train_loader), advance=0) for _, (mri, xls, targets, _) in enumerate(train_loader): _check_control_events(stop_event=stop_event, pause_event=pause_event) mri, xls, targets = _move_batch_to_model_device(model, mri, xls, targets) optimizer.zero_grad() outputs = model((mri, xls)) loss = criterion(outputs, targets) loss.backward() # type: ignore[reportUnknownMemberType] optimizer.step() batch_size = mri.size(0) train_loss += loss.item() * batch_size train_total_samples += batch_size batch_correct, batch_total = _batch_correct_and_total(outputs, targets) train_correct += batch_correct train_total += batch_total train_progress.update(total=len(train_loader), advance=1) train_loss = train_loss / train_total_samples if train_total_samples > 0 else 0.0 model.eval() val_loss = 0.0 correct = 0 total = 0 val_total_samples = 0 test_progress = progress.get_sub_tracker("Validation Batches") test_progress.reset() test_progress.set_title("Validation Batches") test_progress.update(total=len(val_loader), advance=0) with torch.no_grad(): for _, (mri, xls, targets, _) in enumerate(val_loader): _check_control_events(stop_event=stop_event, pause_event=pause_event) mri, xls, targets = _move_batch_to_model_device(model, mri, xls, targets) outputs = model((mri, xls)) loss = criterion(outputs, targets) batch_size = mri.size(0) val_loss += loss.item() * batch_size val_total_samples += batch_size # Calculate accuracy batch_correct, batch_total = _batch_correct_and_total(outputs, targets) correct += batch_correct total += batch_total test_progress.update(total=len(val_loader), advance=1) val_loss = val_loss / val_total_samples if val_total_samples > 0 else 0.0 val_acc = correct / total if total > 0 else 0.0 train_acc = train_correct / train_total if train_total > 0 else 0.0 return train_loss, val_loss, train_acc, val_acc def train_model( model: nn.Module, train_loader: DataLoader[ADNIDataset], val_loader: DataLoader[ADNIDataset], optimizer: torch.optim.Optimizer, criterion: nn.Module, num_epochs: int, output_path: pl.Path, progress: ProgressTracker, log: PipelineLogger, stop_event: Event | None = None, pause_event: Event | None = None, ) -> Tuple[nn.Module, pd.DataFrame]: """ Trains the model using the provided training and validation data loaders. Args: model (nn.Module): The model to train. train_loader (DataLoader[ADNIDataset]): DataLoader for the training dataset. val_loader (DataLoader[ADNIDataset]): DataLoader for the validation dataset. num_epochs (int): Number of epochs to train the model. learning_rate (float): Learning rate for the optimizer. Returns: Result[nn.Module, str]: A Result object containing the trained model or an error message. """ # Record the training history # We record the Epoch, Training Loss, Validation Loss, Training Accuracy, and Validation Accuracy # use a (num_epochs, 4) shape ndarray to store the history before creating the DataArray nhist = np.zeros((num_epochs, 4), dtype=np.float32) # Get new progress bar for epochs epoch_progress = progress.get_sub_tracker("Epochs") epoch_progress.update(total=num_epochs, advance=0) for epoch in range(num_epochs): _check_control_events(stop_event=stop_event, pause_event=pause_event) train_loss, val_loss, train_acc, val_acc = train_epoch( model, train_loader, val_loader, optimizer, criterion, log=log, progress=epoch_progress, stop_event=stop_event, pause_event=pause_event, ) # Update the history nhist[epoch, 0] = train_loss nhist[epoch, 1] = val_loss nhist[epoch, 2] = train_acc nhist[epoch, 3] = val_acc log.info( f"Epoch [{epoch + 1}/{num_epochs}], " f"Train Loss: {train_loss:.4f}, Val Loss: {val_loss:.4f}, " f"Train Acc: {train_acc:.4f}, Val Acc: {val_acc:.4f}" ) # If we are at 25, 50, or 75% of the epochs, save the model if num_epochs > 4: if (epoch + 1) % (num_epochs // 4) == 0: model_save_path = ( output_path / "intermediate_models" / f"model_epoch_{epoch + 1}.pt" ) torch.save(model.state_dict(), model_save_path) log.info(f"Model saved at epoch {epoch + 1}") epoch_progress.update(total=num_epochs, advance=1) # return the trained model and the traning history history = pd.DataFrame( data=nhist.astype(np.float32), columns=["train_loss", "val_loss", "train_acc", "val_acc"], index=np.arange(1, num_epochs + 1), ) return model, history def test_model_bayesian( model: nn.Module, test_loader: DataLoader[ADNIDataset], criterion: nn.Module, get_kl_loss: KLLossFn, progress: ProgressTracker, log: PipelineLogger, stop_event: Event | None = None, pause_event: Event | None = None, ) -> TestMetrics: """ Tests a Bayesian model on the test dataset with KL-augmented loss. """ model.eval() test_loss = 0.0 correct = 0 total = 0 total_samples = 0 test_progress = progress.get_sub_tracker("Testing Batches") test_progress.update(total=len(test_loader), advance=0) with torch.no_grad(): for _, (mri, xls, targets, _) in enumerate(test_loader): _check_control_events(stop_event=stop_event, pause_event=pause_event) mri, xls, targets = _move_batch_to_model_device(model, mri, xls, targets) outputs = model((mri, xls)) data_loss = cast(torch.Tensor, criterion(outputs, targets)) batch_size = mri.size(0) kl_term = get_kl_loss(model) kl_loss = ( kl_term / batch_size if kl_term is not None else torch.tensor(0.0, device=outputs.device) ) loss: torch.Tensor = data_loss + kl_loss test_loss += loss.item() * batch_size total_samples += batch_size batch_correct, batch_total = _batch_correct_and_total(outputs, targets) correct += batch_correct total += batch_total test_progress.update(total=len(test_loader), advance=1) test_loss = test_loss / total_samples if total_samples > 0 else 0.0 test_acc = correct / total if total > 0 else 0.0 return test_loss, test_acc def train_epoch_bayesian( model: nn.Module, train_loader: DataLoader[ADNIDataset], val_loader: DataLoader[ADNIDataset], optimizer: torch.optim.Optimizer, criterion: nn.Module, get_kl_loss: KLLossFn, log: PipelineLogger, progress: ProgressTracker, stop_event: Event | None = None, pause_event: Event | None = None, ) -> TrainMetrics: """ Trains a Bayesian model for one epoch and evaluates on validation data. """ model.train() train_loss = 0.0 train_correct = 0 train_total = 0 train_total_samples = 0 train_progress = progress.get_sub_tracker("Training Batches") train_progress.reset() train_progress.set_title("Training Batches") train_progress.update(total=len(train_loader), advance=0) for _, (mri, xls, targets, _) in enumerate(train_loader): _check_control_events(stop_event=stop_event, pause_event=pause_event) mri, xls, targets = _move_batch_to_model_device(model, mri, xls, targets) optimizer.zero_grad() outputs = model((mri, xls)) data_loss = cast(torch.Tensor, criterion(outputs, targets)) batch_size = mri.size(0) kl_term = get_kl_loss(model) kl_loss = ( kl_term / batch_size if kl_term is not None else torch.tensor(0.0, device=outputs.device) ) loss: torch.Tensor = data_loss + kl_loss loss.backward() # type: ignore[reportUnknownMemberType] optimizer.step() train_loss += loss.item() * batch_size train_total_samples += batch_size batch_correct, batch_total = _batch_correct_and_total(outputs, targets) train_correct += batch_correct train_total += batch_total train_progress.update(total=len(train_loader), advance=1) train_loss = train_loss / train_total_samples if train_total_samples > 0 else 0.0 model.eval() val_loss = 0.0 correct = 0 total = 0 val_total_samples = 0 test_progress = progress.get_sub_tracker("Validation Batches") test_progress.reset() test_progress.set_title("Validation Batches") test_progress.update(total=len(val_loader), advance=0) with torch.no_grad(): for _, (mri, xls, targets, _) in enumerate(val_loader): _check_control_events(stop_event=stop_event, pause_event=pause_event) mri, xls, targets = _move_batch_to_model_device(model, mri, xls, targets) outputs = model((mri, xls)) data_loss = cast(torch.Tensor, criterion(outputs, targets)) batch_size = mri.size(0) kl_term = get_kl_loss(model) kl_loss = ( kl_term / batch_size if kl_term is not None else torch.tensor(0.0, device=outputs.device) ) vloss: torch.Tensor = data_loss + kl_loss val_loss += vloss.item() * batch_size val_total_samples += batch_size batch_correct, batch_total = _batch_correct_and_total(outputs, targets) correct += batch_correct total += batch_total test_progress.update(total=len(val_loader), advance=1) val_loss = val_loss / val_total_samples if val_total_samples > 0 else 0.0 val_acc = correct / total if total > 0 else 0.0 train_acc = train_correct / train_total if train_total > 0 else 0.0 return train_loss, val_loss, train_acc, val_acc def train_model_bayesian( model: nn.Module, train_loader: DataLoader[ADNIDataset], val_loader: DataLoader[ADNIDataset], optimizer: torch.optim.Optimizer, criterion: nn.Module, num_epochs: int, output_path: pl.Path, get_kl_loss: KLLossFn, progress: ProgressTracker, log: PipelineLogger, stop_event: Event | None = None, pause_event: Event | None = None, ) -> Tuple[nn.Module, pd.DataFrame]: """ Trains a Bayesian model with KL-augmented objective. """ nhist = np.zeros((num_epochs, 4), dtype=np.float32) epoch_progress = progress.get_sub_tracker("Epochs") epoch_progress.update(total=num_epochs, advance=0) for epoch in range(num_epochs): _check_control_events(stop_event=stop_event, pause_event=pause_event) train_loss, val_loss, train_acc, val_acc = train_epoch_bayesian( model, train_loader, val_loader, optimizer, criterion, get_kl_loss, log=log, progress=epoch_progress, stop_event=stop_event, pause_event=pause_event, ) nhist[epoch, 0] = train_loss nhist[epoch, 1] = val_loss nhist[epoch, 2] = train_acc nhist[epoch, 3] = val_acc log.info( f"Epoch [{epoch + 1}/{num_epochs}], " f"Train Loss: {train_loss:.4f}, Val Loss: {val_loss:.4f}, " f"Train Acc: {train_acc:.4f}, Val Acc: {val_acc:.4f}" ) if num_epochs > 4: if (epoch + 1) % (num_epochs // 4) == 0: model_save_path = ( output_path / "intermediate_models" / f"model_epoch_{epoch + 1}.pt" ) torch.save(model.state_dict(), model_save_path) log.info(f"Model saved at epoch {epoch + 1}") epoch_progress.update(total=num_epochs, advance=1) history = pd.DataFrame( data=nhist.astype(np.float32), columns=["train_loss", "val_loss", "train_acc", "val_acc"], index=np.arange(1, num_epochs + 1), ) return model, history