Draft for instructor review. You can upload the downloaded notebook to Colab if the first link fails. Use a Python CPU runtime.
Run Setup, then work on the challenge announced in class. Predict a result, adapt the nearby examples to fill the marked lines, and inspect the images and table. Answer the Moodle poll for your own run and be ready to explain one failure.
Allow 5–10 minutes for setup and 15–20 minutes per challenge. The first three challenges compare reconstruction, retrieval and anomaly detection. Two optional challenges explore maps and generation. Compare quality and runtime within each experiment and decide what each metric leaves out.
TODO(instructor): confirm classroom pacing, activity selection and dataset before release.
Run the folded plotting cell above, then run both setup cells.
Enter a birthday month and day without the year, or invent a valid date. Keep it in your notebook. Keep the printed variant for your own run. Moodle asks only which method worked best in each core task.
We use 1,797 offline 8×8 handwritten digits. The textbook experiments use 28×28 MNIST. A row of X has D = 64 pixels, and a row of H has d code coordinates. x_hat is a reconstruction, and y is a digit label. Labels help split the data, define the anomaly scenario and check results. Models fit pixel values.
The fixed split is approximately 60% training, 20% validation and 20% test. Keep the assigned settings for the test poll. Use validation for extra experiments.
from importlib.metadata import PackageNotFoundError, versionimport platformimport shleximport subprocessimport syspackages = ("numpy", "scipy", "scikit-learn", "matplotlib", "torch", "threadpoolctl")def package_versions(): versions = {}for package in packages:try: versions[package] = version(package)except PackageNotFoundError: versions[package] =Nonereturn versionsbefore = package_versions()missing = [name for name, installed in before.items() if installed isNone]if missing: command = [sys.executable, "-m", "pip", "install", *missing]if"google.colab"notin sys.modules:raiseRuntimeError(f"Install missing packages and rerun Setup: {shlex.join(command)}")try: subprocess.run(command, check=True)except (OSError, subprocess.CalledProcessError) as exc:raiseRuntimeError("Package setup failed. Try a fresh Colab session and rerun Setup.") from exc after = package_versions() loaded_names = {"scikit-learn": "sklearn"}ifany(before[name] != after[name] and loaded_names.get(name, name) in sys.modulesfor name in packages):raiseRuntimeError("Restart the session, then rerun Setup to load the installed versions.")print("Python", platform.python_version(), "| CPU")print(package_versions())
Each activity uses a fresh model and task seed. Fit times are one observation. Query and score times are medians of 20 measurements after warm-up, alternating method order. Compare seconds on your own CPU. The tie thresholds are classroom conventions.
incomplete ="I could not obtain a complete result"def run_activity(activity, offset): seed = personal_seed + offset torch.manual_seed(seed) np.random.seed(seed)try: activity(seed)except (AssertionError, TypeError, ValueError, IndexError, RuntimeError) as exc:print(f"{incomplete}. Check the marked lines. {exc}")def timed(operation): start = perf_counter() result = operation()return result, perf_counter() - startdef batch_times(operations): samples = {name: [] for name in operations}for operation in operations.values(): operation() names =list(operations)for repeat inrange(20):for name in (names if repeat %2==0else names[::-1]): _, seconds = timed(operations[name]) samples[name].append(seconds)return {name: float(np.median(times)) for name, times in samples.items()}def report(scores, metric, question=None, tolerance=0, maximize=False, timing=False):assert np.isfinite(list(scores.values())).all(), "A comparison contains a nonfinite result."print(f"\n{metric}")for name, value in scores.items():print(f"{name:34}{value:.6g}") best = (maxif maximize elsemin)(scores.values()) margin = best * tolerance if timing ornot maximize else tolerance winners = [name for name, value in scores.items() ifabs(value - best) <= margin] tie ="Tie / too close to distinguish"if timing else"Tie / practically the same" prefix =f"Moodle Q{question}"if question isnotNoneelse"Result"print(f"{prefix}: {winners[0] iflen(winners) ==1else tie}")class Autoencoder(nn.Module):def__init__(self, width):super().__init__()self.encoder = nn.Sequential(nn.Linear(64, 64), nn.ReLU(), nn.Linear(64, width))self.decoder = nn.Sequential(nn.Linear(width, 64), nn.ReLU(), nn.Linear(64, 64), nn.Sigmoid())def forward(self, x):returnself.decoder(self.encoder(x))def fit_network(model, dataset, seed, objective, epochs=40): batches = DataLoader(dataset, batch_size=128, shuffle=True, generator=torch.Generator().manual_seed(seed)) optimizer = torch.optim.Adam(model.parameters(), lr=0.001)for epoch inrange(epochs):for batch_number, (x, target) inenumerate(batches):assert x.ndim ==2and x.shape == target.shape and x.shape[1] ==64 optimizer.zero_grad() loss = objective(model, x, target)assert torch.isfinite(loss), "Training loss is nonfinite." loss.backward()if epoch ==0and batch_number ==0: gradients = [p.grad for p in model.parameters() if p.grad isnotNone]assertall(torch.isfinite(g).all() for g in gradients)assertany(torch.count_nonzero(g) for g in gradients), "No learning signal." optimizer.step()return model.eval()def mse(model, x, target):return (model(x) - target).square().mean()def clean_ae(images, seed): x = torch.from_numpy(images)return fit_network(Autoencoder(d), TensorDataset(x, x), seed, mse)@torch.inference_mode()def predict(operation, images):return operation(torch.from_numpy(images)).numpy()def fit_pca(images, width):return PCA(n_components=width, svd_solver="full").fit(images)def reconstruct_pca(model, images):return model.inverse_transform(model.transform(images))
Task 1 — Rescue a noisy digit
Predict: Which method will recover the strokes better? Which fits faster?
Complete the first convolution and the linear projection into the code. The shape path is [B, 1, 8, 8] → [B, 8, 8, 8] → [B, 512] → [B, d]. Use a 3×3 kernel with stride 1. The supplied dataset pairs noisy inputs with clean targets, and the decoder restores a flat 64-pixel image.
Compare against PCA fitted on clean training images. Both methods reconstruct the same clipped noisy test images. Score PCA’s raw output and use grayscale limits only for display. The noisy row is a reference.
Moodle Q1: Choose the lower clean-target MSE. Discuss fit time aloud. Errors within 1% and fit times within 10% of the minimum are treated as ties. Point to a digit where the numeric result and your visual judgment differ.
def denoising(seed): rng = np.random.default_rng(seed) noisy = np.clip(X + rng.normal(0, 0.35, X.shape), 0, 1).astype(np.float32) noisy_train, clean_train =map(torch.from_numpy, (noisy[train], X[train]))# Example: a 3×3 convolution from 2 to 4 channels preserves height and width.# nn.Conv2d(2, 4, kernel_size=3, stride=1, padding=1) convolution =None# TODO(student): 1 input channel, 8 output channels# Example: nn.Linear(4 * 8 * 8, 6) maps a flattened 4-channel image to 6 coordinates. projection =None# TODO(student): project the flattened features to dif convolution isNoneor projection isNone:print(incomplete, "Complete the convolution and projection.")returnwith torch.no_grad(): features = convolution(noisy_train[:2].reshape(2, 1, 8, 8))assert features.shape == (2, 8, 8, 8), "Check channels, stride and padding."assert projection(features.flatten(1)).shape == (2, d), "Check the linear dimensions."def fit_denoiser(): model = Autoencoder(d) model.encoder = nn.Sequential( nn.Unflatten(1, (1, 8, 8)), convolution, nn.ReLU(), nn.Flatten(1), projection )return fit_network(model, TensorDataset(noisy_train, clean_train), seed, mse) pca, pca_seconds = timed(lambda: fit_pca(X[train], d)) ae, ae_seconds = timed(fit_denoiser)assert predict(ae.encoder, X[test[:8]]).shape == (8, d) results = {"PCA": reconstruct_pca(pca, noisy[test]),"Denoising autoencoder": predict(ae, noisy[test]), } clean = X[test]print(f"Noisy image MSE (reference): {np.mean((noisy[test] - clean)**2):.6f}") report({name: np.mean((out - clean)**2) for name, out in results.items()},"Test MSE against clean images", 1, tolerance=0.01) report(dict(zip(results, (pca_seconds, ae_seconds))), "Fit seconds", tolerance=0.10, timing=True) rows = [("Clean", clean[:8], None), ("Noisy", noisy[test[:8]], None)] rows += [(name, out[:8], None) for name, out in results.items()] image_rows(rows, "Eight fixed test digits: clean target, noisy input and both reconstructions.")run_activity(denoising, 100)
Task 2 — Find a matching digit
Predict: Does a useful reconstruction code also make a good search key?
Find one nearest training image for each test query. Compare original pixels, PCA codes and autoencoder codes using Euclidean distance in each space. Fit representations on clean training images. A match means the same digit label. The query and gallery have disjoint row IDs.
Select one gallery index per query, count equal digit labels, and turn the count into a top-1 accuracy. The distance matrix has one row per query and one column per gallery image.
Moodle Q2: Choose the most same-digit matches. Discuss query time aloud. A difference of one match or less counts as a tie. Timing includes encoding queries and searching the prepared gallery. Times within 10% of the minimum count as a tie.
Predict: Can an unfamiliar digit be easy to reconstruct?
Digits 0–8 define normal training data. In a mixed test batch of 80 normal images and 20 nines, flag ten images. Fit k-means, PCA and an autoencoder only on normal training rows. Labels define the task and reveal the result after scoring.
Complete the reduction from squared pixel residuals to one score per image. Then order the scores so the first rows are the most suspicious and count actual nines in the flagged set. At ranking time, original row IDs break ties.
Moodle Q3: Choose the most nines among the ten flags. Discuss scoring time aloud. Equal counts are a tie. Times within 10% of the minimum are a tie. The three methods’ raw scores are on different scales.
def reconstruction_scores(x, x_hat): squared = (x - x_hat)**2# Example: squared.mean(axis=0) gives one value per pixel across images.returnNone# TODO(student): reduce the pixel axisdef rank_flags(scores):# Example: np.argsort(scores, kind="stable") orders the smallest scores first.returnNone# TODO(student): descending stable argsortdef count_intruders(flag_labels):# Example: int(np.sum(flag_labels == 2)) counts the twos.returnNone# TODO(student): count actual anomaliesdef anomalies(seed):if reconstruction_scores(X[test[:2]], X[test[:2]]) isNoneor rank_flags(np.array([0., 1.])) isNoneor count_intruders(np.array([9, 2, 9])) isNone:print(incomplete, "Complete the reduction, ranking and count.")returnassert count_intruders(np.array([9, 2, 9])) ==2, "Count only the nines." rng = np.random.default_rng(seed) normal_train = train[y[train] !=9] normal_test, unfamiliar = test[y[test] !=9], test[y[test] ==9] ids = np.r_[rng.choice(normal_test, 80, replace=False), rng.choice(unfamiliar, 20, replace=False)] rng.shuffle(ids)assertlen(np.unique(ids)) ==100and (y[ids] ==9).sum() ==20assert (y[normal_train] !=9).all()# Scores use a shuffled batch. Original IDs break ties at the ranking step. x = X[ids] kmeans = KMeans(10, n_init=10, random_state=seed).fit(X[normal_train]) pca = fit_pca(X[normal_train], d) ae = clean_ae(X[normal_train], seed) operations = {"Distance to k-means centers": lambda: cdist(x, kmeans.cluster_centers_, "sqeuclidean").min(axis=1),"PCA reconstruction error": lambda: reconstruction_scores(x, reconstruct_pca(pca, x)),"Autoencoder reconstruction error": lambda: reconstruction_scores(x, predict(ae, x)), } flags = {} row_order = np.argsort(ids)for name, operation in operations.items(): scores = np.asarray(operation())assert scores.shape == (100,), "Average across pixels to get one score per image." scores = scores[row_order] rank = rank_flags(scores)assert np.array_equal(np.sort(rank), np.arange(100)), "Return 100 row indices."assert np.all(np.diff(scores[rank]) <=0), "Put the highest scores first." flags[name] = ids[row_order[rank[:10]]] report({name: count_intruders(y[flag]) for name, flag in flags.items()},"Nines among ten flags (80 normal / 20 unfamiliar)", 3, maximize=True) report(batch_times(operations), "Score batch seconds, excluding ranking", tolerance=0.10, timing=True) image_rows([(name, X[flag], y[flag]) for name, flag in flags.items()],"The ten highest-scoring images per method with true labels revealed.")run_activity(anomalies, 300)
Extension A — Can you trust this map?
Predict: Which map keeps more of each image’s ten closest neighbors?
Fit two-dimensional PCA and t-SNE maps to the same 500 images. Complete the step that excludes each image from its own neighbor list, then calculate neighbor retention. Use the diagonal of the square distance matrix. Labels color the maps after fitting.
Discuss: Compare PCA with perplexity-30 t-SNE. Report average original-neighbor retention and map construction time. Differences of at most 0.01 retention and times within 10% of the minimum are ties. Compare the extra perplexity-10 view visually. What could you infer about neighboring images? What would you avoid inferring about gaps between islands?
def map_neighbors(coordinates): distances = cdist(coordinates, coordinates, "sqeuclidean")# Example: setting a candidate distance to np.inf keeps it out of the nearest ten. self_distance =None# TODO(student): value to put on the diagonalif self_distance isNone:returnNone np.fill_diagonal(distances, self_distance) neighbors = np.argsort(distances, axis=1, kind="stable")[:, :10]assertnot np.any(neighbors == np.arange(len(coordinates))[:, None]), "Exclude each row itself."return neighborsdef neighbor_overlap(original, mapped):# Example: np.intersect1d([0, 1], [1, 2]) returns array([1]).returnNone# TODO(student): fraction of original neighbors retaineddef maps(seed): ids, _ = train_test_split(train, train_size=500, stratify=y[train], random_state=1010) pixels = X[ids] original = map_neighbors(pixels) probe = neighbor_overlap(np.array([0, 1]), np.array([1, 2]))if original isNoneor probe isNone:print(incomplete, "Complete self-exclusion and neighbor overlap.")returnassert probe ==0.5, "One of two original neighbors is retained." pca_map, pca_seconds = timed(lambda: PCA(2, svd_solver="full").fit_transform(pixels)) tsne_map, tsne_seconds = timed(lambda: tsne(30, seed).fit_transform(pixels)) views = {"PCA map": pca_map, "t-SNE map (perplexity 30)": tsne_map} neighbors = {name: map_neighbors(coords) for name, coords in views.items()} retention = { name: np.mean([neighbor_overlap(a, b) for a, b inzip(original, mapped)])for name, mapped in neighbors.items() } report(retention, "Fraction of ten original neighbors retained", tolerance=0.01, maximize=True) report(dict(zip(views, (pca_seconds, tsne_seconds))), "Map construction seconds", tolerance=0.10, timing=True) views["t-SNE map (perplexity 10)"] = tsne(10, seed).fit_transform(pixels) plot_maps(views, y[ids])for i in (0, 1, 2): rows = [("Query + pixels", pixels[np.r_[i, original[i, :4]]], None)] rows += [(name, pixels[np.r_[i, near[i, :4]]], None) for name, near in neighbors.items()] image_rows(rows, f"Image {ids[i]} at left, followed by four neighbors per space.")run_activity(maps, 400)
Extension B — Reconstruct or generate?
Predict: Will the better reconstructor also decode random codes into more recognizable digits?
Complete the sampled code and the sum of KL contributions for each image. The encoder predicts a mean and log variance. Convert log variance to a standard deviation and scale the supplied standard-normal noise before shifting by the mean.
The VAE minimizes Gaussian reconstruction negative log likelihood (pixel standard deviation 0.25) plus KL to a standard normal. Sum each term over its features, then average over the batch. The autoencoder trains with pixel MSE. Compare the two using the same test MSE, decoding the VAE encoder mean for a repeatable diagnostic.
Discuss: Choose lower test MSE (within 1% is a tie), then judge which random-code grid contains more recognizable digits. Both grids use the same 16 codes. Images show decoder means. The AE did not learn a standard-normal prior.
def sample_code(mu, logvar, epsilon):# Example: mean + std * noise shifts and scales standard-normal noise. std =None# TODO(student): convert log variance to standard deviationreturnNoneif std isNoneelse mu + std * epsilonclass VariationalAutoencoder(nn.Module):def__init__(self, width):super().__init__()self.encoder = nn.Sequential(nn.Linear(64, 64), nn.ReLU())self.mu = nn.Linear(64, width)self.logvar = nn.Linear(64, width)self.decoder = nn.Sequential(nn.Linear(width, 64), nn.ReLU(), nn.Linear(64, 64), nn.Sigmoid())def encode(self, x): h =self.encoder(x)returnself.mu(h), self.logvar(h)def kl_to_normal(mu, logvar): per_coordinate =0.5* (mu.square() + logvar.exp() -1- logvar)# Example: per_coordinate.sum(dim=0) gives one total per latent coordinate.returnNone# TODO(student): sum the KL coordinates for each imagedef vae_objective(model, x, target): mu, logvar = model.encode(x) z_sample = sample_code(mu, logvar, torch.randn_like(mu))assert z_sample.shape == mu.shape x_hat = model.decoder(z_sample) sd =0.25 nll = (0.5* ((target - x_hat) / sd).square() + math.log(sd * math.sqrt(2* math.pi))).sum(dim=1)return (nll + kl_to_normal(mu, logvar)).mean()def generation(seed): zero, one = torch.zeros(2, d), torch.ones(2, d) sample = sample_code(zero, zero, one)if sample isNoneor kl_to_normal(zero, zero) isNone:print(incomplete, "Complete the sampling and KL reduction.")returnassert torch.allclose(sample, one), "At zero mean and unit variance the sample equals the noise."assert torch.allclose(sample_code(one, one * math.log(4), one), 3* one), "Check the standard deviation."assert torch.equal(kl_to_normal(zero, zero), torch.zeros(2))assert torch.allclose(kl_to_normal(one, zero), torch.full((2,), d /2)), "Check the sum axis." x = torch.from_numpy(X[train]) ae = clean_ae(X[train], seed) vae = fit_network(VariationalAutoencoder(d), TensorDataset(x, x), seed, vae_objective) clean = X[test] outputs = {"Autoencoder": predict(ae, clean),"Variational autoencoder": predict(lambda x: vae.decoder(vae.encode(x)[0]), clean), } report({name: np.mean((out - clean)**2) for name, out in outputs.items()}, "Test MSE", tolerance=0.01) rows = [("Clean", clean[:8], None)] + [(name, out[:8], None) for name, out in outputs.items()] image_rows(rows, "Eight fixed test digits and their deterministic reconstructions.") codes = np.random.default_rng(seed).standard_normal((16, d)).astype(np.float32) grids = [("Autoencoder grid", predict(ae.decoder, codes)), ("Variational autoencoder grid", predict(vae.decoder, codes))] image_rows([(name if start ==0else"", out[start:start +8], None)for name, out in grids for start in (0, 8)],"Judge the two 16-image grids decoded from the same normal codes.")run_activity(generation, 500)
Discuss your results
Choose a method for one job. Give one result that supports your choice, one image where it fails, and one cost you would consider before using it. Compare votes as observations from these runs, including ties and incomplete results.
TODO(instructor): confirm Moodle access and reporting, gradebook treatment, classroom timing and the latest Colab runtime before release.