Step 1: A Dataset and DataLoaders
Northgate Post's sorting office scans handwritten postcodes and wants a network that reads the digits. You have 1,797 scanned digits, each an 8x8 grid of pixel values from 0 to 16, split into training, validation and test.
PyTorch feeds a network through two pieces:
- A Dataset answers two questions: how many items there are, and what item i is.
- A DataLoader groups items into shuffled batches.
1. Write DigitsDataset: flatten each image to 64 values scaled to 0-1 (float32), and return it with its
label as an int64 tensor.
2. Write make_loaders(): batches of 32 for training, shuffled from a seeded generator; batches of 256 for
validation and test, unshuffled.
3. Run it and look at a few digits from a training batch.
digits.py, the file you edit103 lines
"""Read handwritten postcode digits for Northgate Post's sorting office: a small network, trained properly."""
import copy
import random
import numpy as np
import torch
from sklearn.datasets import load_digits
from sklearn.model_selection import train_test_split
from torch import nn
from torch.utils.data import DataLoader, Dataset
torch.set_num_threads(1) # the lab has one CPU; more threads only fight over it
def split_data():
"""The 1,797 scanned 8x8 digits split into train (60%), validation (20%) and test (20%), stratified,
random_state=0. Returns ((X_train, y_train), (X_val, y_val), (X_test, y_test)) as numpy arrays; X is
(n, 8, 8) with pixel values 0-16."""
d = load_digits()
Xa, Xte, ya, yte = train_test_split(d.images, d.target, test_size=0.2, stratify=d.target, random_state=0)
Xtr, Xva, ytr, yva = train_test_split(Xa, ya, test_size=0.25, stratify=ya, random_state=0)
return (Xtr, ytr), (Xva, yva), (Xte, yte)
# ---------- Step 1: a Dataset and DataLoaders ----------
class DigitsDataset(Dataset):
"""One item = (pixels, label): pixels a float32 tensor of 64 values scaled to 0-1 (divide by 16), label an
int64 tensor."""
def __init__(self, images, labels):
# TODO (Step 1): self.x = the images flattened to (n, 64), divided by 16, as a float32 tensor;
# self.y = the labels as an int64 (torch.long) tensor.
raise NotImplementedError("Step 1: write DigitsDataset.__init__()")
def __len__(self):
raise NotImplementedError("Step 1: write DigitsDataset.__len__()")
def __getitem__(self, i):
raise NotImplementedError("Step 1: write DigitsDataset.__getitem__()")
def make_loaders(batch_size=32, seed=0):
"""(train_loader, val_loader, test_loader). The training loader shuffles, using a torch.Generator seeded with
`seed`; the other two keep their order and use batches of 256."""
# TODO (Step 1): split_data(), then a DataLoader per split: training batch_size=batch_size, shuffle=True,
# generator=torch.Generator().manual_seed(seed); validation and test batch_size=256, no shuffle.
raise NotImplementedError("Step 1: write make_loaders()")
# ---------- Step 2: the model and one epoch of training ----------
class MLP(nn.Module):
"""64 pixels -> Linear(64, hidden) -> ReLU -> Dropout(dropout) -> Linear(hidden, hidden) -> ReLU ->
Dropout(dropout) -> Linear(hidden, 10): one score (logit) per digit."""
def __init__(self, hidden=256, dropout=0.2):
super().__init__()
raise NotImplementedError("MLP arrives in Step 2")
def forward(self, x):
raise NotImplementedError("MLP arrives in Step 2")
def train_one_epoch(model, loader, optimizer, loss_fn):
"""One pass over the loader in training mode: for every batch clear the gradients, predict, compute the loss,
backpropagate and step. Returns the mean loss per example (each batch weighted by its size), as a float."""
raise NotImplementedError("train_one_epoch() arrives in Step 2")
# ---------- Step 3: evaluation ----------
def evaluate(model, loader, loss_fn):
"""Mean loss per example and accuracy over the loader, in evaluation mode (dropout off) and without building
a gradient graph: {"loss": float, "accuracy": float}."""
raise NotImplementedError("evaluate() arrives in Step 3")
# ---------- Step 4: the full loop, with early stopping ----------
def fit(model, train_loader, val_loader, epochs=60, patience=8, lr=1e-3):
"""Train with Adam(lr) and cross-entropy for up to `epochs` epochs, evaluating on the validation loader after
each. Keep a copy of the weights (state_dict) from the epoch with the lowest validation loss; stop when
`patience` epochs in a row have not beaten it; load the best weights back into the model before returning.
Returns the history: one {"epoch", "train_loss", "val_loss", "val_accuracy"} per epoch run (epoch from 1)."""
raise NotImplementedError("fit() arrives in Step 4")
# ---------- Step 5: reproducible, saved, tested once ----------
def seed_everything(seed):
"""Seed Python's random, numpy and torch."""
raise NotImplementedError("seed_everything() arrives in Step 5")
def train_run(seed=0):
"""seed_everything(seed), build MLP(), make_loaders(seed=seed), fit() it. Returns (model, history, loaders)."""
raise NotImplementedError("train_run() arrives in Step 5")
def save(model, path):
"""Save the model's weights (its state_dict) to path."""
raise NotImplementedError("save() arrives in Step 5")
def load_model(path):
"""A fresh MLP() with the weights from path, in evaluation mode."""
raise NotImplementedError("load_model() arrives in Step 5")try_it.py