""" contains functions for training and testing a pytorch model """ import torch from tqdm.auto import tqdm from typing import Dict, List, Tuple # from torch.utils.tensorboard.writer import SummaryWriter def train_step(model: torch.nn.Module, dataloader: torch.utils.data.DataLoader, loss_fn: torch.nn.Module, optimizer: torch.optim.Optimizer, device: torch.device) -> Tuple[float, float]: """Trains a pytorch model for a single epoch turns a target model to training mode then runs through all of the required training steps (forward pass, loss calculation, optimizer step). Args: model: pytorch model dataloader: dataloader insatnce for the model to be trained on loss_fn: pytorch loss function to calculate loss optimizer: pytorch optimizer to help minimize the loss function device: target device returns: a tuple of training loss and training accuracy metrics in the form (train_loss, train_accuracy) """ # put the model into training mode model.train() # setup train loss and train accuracy train_loss, train_accuracy = 0, 0 # loop through data laoder batches for batch, (X, y) in enumerate(dataloader): # send data to target device X, y = X.to(device), y.to(device) # forward pass logits = model(X) # calculate loss and accumulate loss loss = loss_fn(logits, y) train_loss += loss # optimizer zero grad optimizer.zero_grad() # loss backward loss.backward() # optimizer step optimizer.step() # calculate and accumulate accuracy metric across all batches preds = torch.softmax(logits, dim=-1).argmax(dim=-1) train_accuracy += (preds == y).sum().item()/len(preds) # adjust metrics to get average loss and accuracy per batch train_loss /= len(dataloader) train_accuracy /= len(dataloader) return train_loss, train_accuracy def test_step(model: torch.nn.Module, dataloader: torch.utils.data.DataLoader, loss_fn: torch.nn.Module, device: torch.device) -> Tuple[float, float]: """Tests a pytorch model for a single epoch Turns a target model to eval mode and then performs a forward pass on a testing dataset. Args: model: pytorch model dataloader: dataloader insatnce for the model to be tested on loss_fn: loss function to calculate loss (errors) device: target device to compute on returns: A tuple of testing loss and testing accuracy metrics. In the form (test_loss, test_accuracy) """ # put the model in eval mode model.eval() # setup test loss and test accuracy test_loss, test_accuracy = 0, 0 # turn on inference mode with torch.inference_mode(): # loop through all batches for X, y in dataloader: # send data to target device X, y = X.to(device), y.to(device) # forward pass logits = model(X) # calculate and accumulate loss loss = loss_fn(logits, y) test_loss += loss.item() # calculate and accumulate accuracy test_preds = torch.softmax(logits, dim=-1).argmax(dim=-1) test_accuracy += ((test_preds == y).sum().item()/len(test_preds)) # adjust metrics to get average loss and accuracy per batch test_loss /= len(dataloader) test_accuracy /= len(dataloader) return test_loss, test_accuracy def train(model: torch.nn.Module, train_dataloader: torch.utils.data.DataLoader, test_dataloader: torch.utils.data.DataLoader, optimizer: torch.optim.Optimizer, loss_fn: torch.nn.Module, epochs: int, device: torch.device, writer: torch.utils.tensorboard.writer.SummaryWriter) -> Dict[str, List]: """Trains and tests pytorch model passes a target model through train_step() and test_step() functions for a number of epochs, training and testing the model in the same epoch loop. calculates, prints and stores evaluation metric throughout. Args: model: pytorch model train_dataloader: DataLoader instance for the model to be trained on test_dataloader: DataLoader instance for the model to be tested on optimizer: pytorch optimizer loss_fn: pytorch loss function epochs: integer indicating how many epochs to train for device: target device to compute on returns: A dictionaru of training and testing loss as well as training and testing accuracy metrics. Each metric has a value in a list for each epoch. In the form: {train_loss: [...], train_acc: [...], test_loss: [...], test_acc: [...]} """ # create an empty dictionary results = { "train_loss": [], "train_acc": [], "test_loss": [], "test_acc": [] } # loop through training and testing steps for a number of epochs for epoch in tqdm(range(epochs)): train_loss, train_acc = train_step(model=model, dataloader=train_dataloader, loss_fn=loss_fn, optimizer=optimizer, device=device) test_loss, test_acc = test_step(model=model, dataloader=test_dataloader, loss_fn=loss_fn, device=device) if epoch % 1 == 0: print( f"Epoch: {epoch+1} | " f"train_loss: {train_loss:.4f} | " f"train_acc: {train_acc:.4f} | " f"test_loss: {test_loss:.4f} | " f"test_acc: {test_acc:.4f}" ) # update results dictionary results["train_loss"].append(train_loss.item()) results["train_acc"].append(train_acc) results["test_loss"].append(test_loss) results["test_acc"].append(test_acc) if writer: # NEW: EXPERIMENT TRACKING # add loss to SummaryWriter writer.add_scalars(main_tag="Loss", tag_scalar_dict={"train loss": train_loss, "test loss": test_loss}, global_step=epoch) # add accuracy to SummaryWriter writer.add_scalars(main_tag="Accuracy", tag_scalar_dict={"train acc": train_acc, "test acc": test_acc}, global_step=epoch) # track the pytorch model architecture writer.add_graph(model=model, input_to_model=torch.randn(size=(32, 3, 224, 224)).to(device)) writer.close() # END SummaryWriter tracking process # return the filled results dictionaru return results