Project: EMNIST letter classifier

Follow a letter from a one-channel image through convolution to 26 class scores, then inspect the model's mistakes.

The question

Why does keeping an image as a grid usually give a model more useful structure than flattening it immediately?

What to remember

A dense baseline can classify pixels, but it loses the explicit neighbor relationship between strokes. A convolutional model reuses small local detectors across the image before turning the result into 26 logits.

Key code

self.features = nn.Sequential(
    nn.Conv2d(1, 32, 3, padding=1),
    nn.BatchNorm2d(32), nn.ReLU(), nn.MaxPool2d(2),
    nn.Conv2d(32, 64, 3, padding=1),
    nn.BatchNorm2d(64), nn.ReLU(), nn.MaxPool2d(2),
    nn.AdaptiveAvgPool2d((1, 1)),
)
self.classifier = nn.Linear(64, classes)

The convolution blocks preserve the two-dimensional view while pooling reduces spatial size. AdaptiveAvgPool2d((1, 1)) creates a stable 64-feature boundary before classification.

Trace it: an EMNIST batch enters as [N, 1, 28, 28]. The two convolutions grow channels to 32 and then 64; the two pools shrink 28 → 14 → 7. Adaptive pooling gives [N, 64, 1, 1], flattening gives [N, 64], and the classifier returns [N, 26]. Explain each change before running the full script.

Evidence and visual

The included full project downloads the public EMNIST Letters split through TorchVision, trains this CNN, and saves its own predictions, confusion matrix, curves, checkpoint, and JSON metrics. The shape map remains useful before any dataset is downloaded.

Reproduced CPU run

Historical evidence: the images below come from the original CPU configuration, which evaluated test at each epoch: seed 42, 4,000 training examples, 1,000 test examples, and three epochs. It reached 24.4% test accuracy. This intentionally small configuration demonstrates the full workflow and its artifacts; it is not presented as a strong handwriting benchmark.

Letter batch[8,1,28,28]8 grayscale images

Conv + ReLU[8,32,28,28]32 feature maps

Pool[8,32,14,14]halve spatial size

Conv + pool[8,64,7,7]64 feature maps

Adaptive avg + flatten[8,64]one value per channel

Classifier[8,26]26 raw letter scores

Illustration: shapes and operations, not measured model performance.

Interactive check

EMNIST spatial-size tracer

Input width and height 2× pooling blocks Trace shape

Run it yourself

python examples/emnist_model.py

The default small run downloads data when needed, splits a 4,000-image pool into 3,200 training and 800 validation images, trains for three epochs, and writes artifacts to artifacts/emnist/. It uses CUDA if available; add --device cpu to force CPU. For a longer GPU run:

python examples/emnist_model.py --full --device cuda

Use --smoke-test to verify model shapes without downloading data.

Validation and final test

The maintained script now splits the training pool with a fixed seed (--validation-fraction 0.2). It chooses the minimum validation-loss checkpoint, restores those weights, then evaluates the official test split at the end. The saved report records disjoint source indices and best_epoch; the checkpoint records normalization and class order. Early stopping would end training earlier and is a separate decision.

The default 4,000-image training pool becomes 3,200 training and 800 validation images; the separate test subset remains 1,000 images. The retained figures and 24.4% result above describe the original three-epoch configuration, which evaluated test during training. They are historical evidence and have not been relabeled as validation results or regenerated by this change.

Complete source

Open the maintained runnable script

"""Train a reproducible EMNIST letter classifier with public TorchVision data.

The default run is deliberately small: it downloads EMNIST Letters when
needed, trains on a fixed subset, and saves figures under artifacts/. Device
selection is automatic; pass --device cpu to require CPU.
Use --full --device cuda for a longer run on a compatible GPU.
"""

from __future__ import annotations

import argparse
import json
import random
import string
import time
from pathlib import Path

try:
    from .validation_patterns import split_indices, BestState
except ImportError:  # direct script execution
    from validation_patterns import split_indices, BestState

import matplotlib.pyplot as plt
import torch
from torch import nn
from torch.utils.data import DataLoader, Subset
from torchvision import datasets, transforms


ROOT = Path(__file__).resolve().parents[1]
LETTERS = list(string.ascii_uppercase)


class DenseLetterClassifier(nn.Module):
    def __init__(self, classes: int = 26) -> None:
        super().__init__()
        self.network = nn.Sequential(
            nn.Flatten(), nn.Linear(28 * 28, 256), nn.ReLU(), nn.Linear(256, classes)
        )

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        return self.network(x)


class ConvolutionalLetterClassifier(nn.Module):
    def __init__(self, classes: int = 26) -> None:
        super().__init__()
        self.features = nn.Sequential(
            nn.Conv2d(1, 32, 3, padding=1), nn.BatchNorm2d(32), nn.ReLU(), nn.MaxPool2d(2),
            nn.Conv2d(32, 64, 3, padding=1), nn.BatchNorm2d(64), nn.ReLU(), nn.MaxPool2d(2),
            nn.AdaptiveAvgPool2d((1, 1)),
        )
        self.classifier = nn.Linear(64, classes)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        return self.classifier(self.features(x).flatten(1))


def set_seed(seed: int) -> None:
    random.seed(seed)
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)


def device_for(requested: str) -> torch.device:
    if requested == "auto":
        return torch.device("cuda" if torch.cuda.is_available() else "cpu")
    device = torch.device(requested)
    if device.type == "cuda" and not torch.cuda.is_available():
        raise RuntimeError("CUDA was requested but is not available. Use --device cpu or --device auto.")
    return device


def deterministic_subset(dataset, limit: int | None, seed: int) -> Subset | datasets.EMNIST:
    if limit is None or limit >= len(dataset):
        return dataset
    indices = torch.randperm(len(dataset), generator=torch.Generator().manual_seed(seed))[:limit].tolist()
    return Subset(dataset, indices)


def load_emnist(data_root: Path, *, train: bool, transform, target_transform):
    """Load EMNIST and retry a Windows cleanup race after a completed download."""
    options = {
        "split": "letters",
        "train": train,
        "download": True,
        "transform": transform,
        "target_transform": target_transform,
    }
    try:
        return datasets.EMNIST(data_root, **options)
    except PermissionError:
        # TorchVision has already extracted the files in this case; a short retry
        # lets a transient OneDrive/antivirus file handle clear on Windows.
        time.sleep(1)
        return datasets.EMNIST(data_root, **options)


def train_epoch(model, loader, optimizer, loss_fn, device: torch.device) -> tuple[float, float]:
    model.train()
    total_loss = total_correct = total_count = 0
    for images, labels in loader:
        images, labels = images.to(device), labels.to(device)
        optimizer.zero_grad(set_to_none=True)
        logits = model(images)
        loss = loss_fn(logits, labels)
        loss.backward()
        optimizer.step()
        total_loss += loss.item() * labels.size(0)
        total_correct += (logits.argmax(1) == labels).sum().item()
        total_count += labels.size(0)
    return total_loss / total_count, total_correct / total_count


@torch.inference_mode()
def evaluate(model, loader, loss_fn, device: torch.device) -> tuple[float, float, torch.Tensor, torch.Tensor]:
    model.eval()
    total_loss = total_correct = total_count = 0
    labels_all, predictions_all = [], []
    for images, labels in loader:
        logits = model(images.to(device))
        total_loss += loss_fn(logits, labels.to(device)).item() * labels.size(0)
        predictions = logits.argmax(1).cpu()
        total_correct += (predictions == labels).sum().item()
        total_count += labels.size(0)
        labels_all.append(labels)
        predictions_all.append(predictions)
    return total_loss / total_count, total_correct / total_count, torch.cat(labels_all), torch.cat(predictions_all)


def save_figures(model, loader, labels, predictions, history, output: Path, device: torch.device) -> None:
    output.mkdir(parents=True, exist_ok=True)
    images, actual = next(iter(loader))
    with torch.inference_mode():
        predicted = model(images.to(device)).argmax(1).cpu()
    figure, axes = plt.subplots(3, 4, figsize=(8, 6))
    for axis, image, truth, guess in zip(axes.flat, images[:12], actual[:12], predicted[:12]):
        axis.imshow(image.squeeze(0).mul(0.5).add(0.5), cmap="gray")
        axis.set_title(f"{LETTERS[truth]} → {LETTERS[guess]}", color="#1b7f3a" if truth == guess else "#b23a48")
        axis.axis("off")
    figure.suptitle("EMNIST predictions from this run")
    figure.tight_layout()
    figure.savefig(output / "emnist-predictions.png", dpi=160)
    plt.close(figure)

    confusion = torch.zeros(26, 26, dtype=torch.int64)
    for truth, guess in zip(labels, predictions):
        confusion[truth, guess] += 1
    figure, axis = plt.subplots(figsize=(9, 7))
    image = axis.imshow(confusion, cmap="Blues")
    axis.set(title="EMNIST confusion matrix from this run", xlabel="Predicted letter", ylabel="True letter")
    axis.set_xticks(range(26), LETTERS, fontsize=7)
    axis.set_yticks(range(26), LETTERS, fontsize=7)
    figure.colorbar(image, ax=axis, label="count")
    figure.tight_layout()
    figure.savefig(output / "emnist-confusion-matrix.png", dpi=160)
    plt.close(figure)

    figure, axes = plt.subplots(1, 2, figsize=(9, 3.4))
    axes[0].plot(history["train_loss"], label="train")
    axes[0].plot(history["validation_loss"], label="validation")
    axes[0].set(title="Loss", xlabel="epoch"); axes[0].legend()
    axes[1].plot(history["train_accuracy"], label="train")
    axes[1].plot(history["validation_accuracy"], label="validation")
    axes[1].set(title="Accuracy", xlabel="epoch"); axes[1].legend()
    figure.tight_layout()
    figure.savefig(output / "emnist-training-curves.png", dpi=160)
    plt.close(figure)


def run(args: argparse.Namespace) -> dict[str, object]:
    set_seed(args.seed)
    device = device_for(args.device)
    transform = transforms.Compose([
        transforms.ToTensor(), transforms.Normalize((0.5,), (0.5,)),
    ])
    target = lambda value: value - 1  # EMNIST Letters labels are 1–26; CrossEntropyLoss expects 0–25.
    data_root = Path(args.data_dir)
    train_data = load_emnist(data_root, train=True, transform=transform, target_transform=target)
    test_data = load_emnist(data_root, train=False, transform=transform, target_transform=target)
    train_limit = None if args.full else args.train_limit
    test_limit = None if args.full else args.test_limit
    train_data = deterministic_subset(train_data, train_limit, args.seed)
    test_data = deterministic_subset(test_data, test_limit, args.seed + 1)
    train_indices, val_indices = split_indices(len(train_data), args.validation_fraction, args.seed)
    # The deterministic preprocessing can be shared; only the indices differ.
    val_data = Subset(train_data, val_indices)
    source_indices = train_data.indices if isinstance(train_data, Subset) else range(len(train_data))
    split_record = {"training_source_indices": [source_indices[i] for i in train_indices],
                    "validation_source_indices": [source_indices[i] for i in val_indices]}
    train_data = Subset(train_data, train_indices)
    loader_args = {"batch_size": args.batch_size, "num_workers": 0, "pin_memory": device.type == "cuda"}
    train_loader = DataLoader(train_data, shuffle=True, **loader_args)
    test_loader = DataLoader(test_data, shuffle=False, **loader_args)
    val_loader = DataLoader(val_data, shuffle=False, **loader_args)
    model = ConvolutionalLetterClassifier().to(device)
    optimizer = torch.optim.AdamW(model.parameters(), lr=args.learning_rate, weight_decay=1e-4)
    loss_fn = nn.CrossEntropyLoss()
    epochs = args.full_epochs if args.full else args.epochs
    history = {key: [] for key in ("train_loss", "train_accuracy", "validation_loss", "validation_accuracy")}
    best = BestState()
    for epoch_index in range(epochs):
        train_loss, train_accuracy = train_epoch(model, train_loader, optimizer, loss_fn, device)
        val_loss, val_accuracy, _, _ = evaluate(model, val_loader, loss_fn, device)
        history["train_loss"].append(train_loss); history["train_accuracy"].append(train_accuracy)
        history["validation_loss"].append(val_loss); history["validation_accuracy"].append(val_accuracy)
        best.consider(model, val_loss, epoch_index + 1)
    best.restore(model)
    test_loss, test_accuracy, labels, predictions = evaluate(model, test_loader, loss_fn, device)
    output = Path(args.output_dir)
    save_figures(model, test_loader, labels, predictions, history, output, device)
    torch.save({"model_state": model.state_dict(), "classes": LETTERS, "history": history, "best_epoch": best.epoch, "normalization": {"mean": [0.5], "std": [0.5]}, "split": split_record}, output / "emnist-model.pt")
    summary = {"device": str(device), "train_examples": len(train_data), "test_examples": len(test_data), "epochs": epochs, "validation_examples": len(val_data), "best_epoch": best.epoch, "selection": "minimum validation loss", "split": split_record, "test_loss": test_loss, "test_accuracy": test_accuracy}
    (output / "emnist-metrics.json").write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8")
    return summary


def smoke_test() -> dict[str, int]:
    torch.manual_seed(42)
    batch = torch.randn(8, 1, 28, 28)
    dense, cnn = DenseLetterClassifier(), ConvolutionalLetterClassifier()
    assert dense(batch).shape == cnn(batch).shape == (8, 26)
    return {"dense_parameters": sum(p.numel() for p in dense.parameters()), "cnn_parameters": sum(p.numel() for p in cnn.parameters())}


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--data-dir", default=ROOT / "data" / "emnist", type=Path)
    parser.add_argument("--output-dir", default=ROOT / "artifacts" / "emnist", type=Path)
    parser.add_argument("--epochs", default=3, type=int)
    parser.add_argument("--full-epochs", default=12, type=int)
    parser.add_argument("--train-limit", default=4000, type=int)
    parser.add_argument("--test-limit", default=1000, type=int)
    parser.add_argument("--batch-size", default=128, type=int)
    parser.add_argument("--learning-rate", default=1e-3, type=float)
    parser.add_argument("--seed", default=42, type=int)
    parser.add_argument("--validation-fraction", type=float, default=0.2)
    parser.add_argument("--device", choices=("auto", "cpu", "cuda"), default="auto")
    parser.add_argument("--full", action="store_true", help="Use the complete dataset and the longer epoch budget.")
    parser.add_argument("--smoke-test", action="store_true", help="Check shapes without downloading data.")
    return parser.parse_args()


if __name__ == "__main__":
    arguments = parse_args()
    print(json.dumps(smoke_test() if arguments.smoke_test else run(arguments), indent=2))

The script imports the shared checkpoint/split helper from examples/validation_patterns.py; keep both files when running outside a full checkout.

Shared deterministic split and best-checkpoint helper

"""Seeded split and best-state selection shared by the image demonstrations."""
import copy
import torch


def split_indices(total, fraction=0.2, seed=42):
    if total < 2 or not 0 < fraction < 1:
        raise ValueError("Need at least two samples and a validation fraction between 0 and 1")
    order = torch.randperm(total, generator=torch.Generator().manual_seed(seed)).tolist()
    count = min(total - 1, max(1, round(total * fraction)))
    return order[count:], order[:count]


class BestState:
    """Retain an independent snapshot, minimizing validation loss."""
    def __init__(self):
        self.loss = float("inf")
        self.epoch = None
        self.state = None

    def consider(self, model, loss, epoch):
        if loss < self.loss:
            self.loss, self.epoch = float(loss), epoch
            self.state = copy.deepcopy(model.state_dict())

    def restore(self, model):
        if self.state is None:
            raise ValueError("No finite validation result was retained")
        model.load_state_dict(self.state)