Which bin does it go in? Build a recycling sorter with transfer learning¶
ITE Higher Nitec in AI Applications: Deep Learning, Computer Vision, Responsible AI
In this lab you teach a computer to look at a photo of waste and say what it is, and so which bin it belongs in. You will not train a deep network from nothing. You will borrow one that has already learned to see from more than a million photos, and teach it one new layer. This is called transfer learning.
The problem. Singapore has one landfill, on Pulau Semakau, and it is expected to be full by about 2035. Recycling only helps if the right things go into the blue bin, clean and empty. One dirty item can spoil a whole bin.
The data. RealWaste: 4,752 colour photos of real waste, in nine classes. Single, S., Iranmanesh, S. and Raad, R. (2023), RealWaste, UCI Machine Learning Repository, DOI 10.24432/C5SS4G, licence CC BY 4.0. The photos were taken at a landfill in Australia.
Rule for the whole lab: never photograph people, faces, name cards or addresses for a model. In the web demo your photo never leaves your phone.
1. Hook: you are the model¶
You will see nine photos below, one from each class. For each, decide: blue recycling bin, or general waste? Write your answers down. At the end of the lab the model answers too.
import os
import io
import glob
import json
import zipfile
import urllib.request
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from PIL import Image
UCI_URL = "https://archive.ics.uci.edu/static/public/908/realwaste.zip"
# On the lab machine the photos are already here; in Colab they are downloaded (about 660 MB).
LOCAL_ROOT = "../ml/data/raw/realwaste/extracted/realwaste-main/RealWaste"
COLAB_ROOT = "/content/realwaste/realwaste-main/RealWaste"
def data_root():
if os.path.isdir(LOCAL_ROOT):
return LOCAL_ROOT
if not os.path.isdir(COLAB_ROOT):
archive = zipfile.ZipFile(io.BytesIO(urllib.request.urlopen(UCI_URL).read()))
archive.extractall("/content/realwaste")
return COLAB_ROOT
ROOT = data_root()
CLASSES = sorted(os.listdir(ROOT))
CLASSES = [name for name in CLASSES if os.path.isdir(os.path.join(ROOT, name))]
photos = pd.DataFrame([{"path": path, "label": name} for name in CLASSES for path in sorted(glob.glob(os.path.join(ROOT, name, "*.jpg")))])
print(len(photos), "photos in", len(CLASSES), "classes:", ", ".join(CLASSES))
4752 photos in 9 classes: Cardboard, Food Organics, Glass, Metal, Miscellaneous Trash, Paper, Plastic, Textile Trash, Vegetation
fig, axes = plt.subplots(1, len(CLASSES), figsize=(18, 2.6))
for axis, name in zip(axes, CLASSES):
axis.imshow(Image.open(photos[photos.label == name].path.iloc[0]))
axis.set_title(name, fontsize=9)
axis.axis("off")
plt.show()
2. Explore the data¶
How many photos of each class? If one class is much bigger, a lazy model can score well by always guessing it. Check before you train.
counts = photos["label"].value_counts()
counts.plot.barh(figsize=(6, 3), color="#4c78a8", title="Photos per class")
plt.show()
print(f"Biggest class: {counts.index[0]}, {counts.iloc[0] / len(photos):.0%} of all photos")
print("Photo size:", Image.open(photos.path.iloc[0]).size)
Biggest class: Plastic, 19% of all photos Photo size: (524, 524)
3. Split: train, check, test¶
Three groups, chosen at random but with the same mix of classes in each (stratified):
- Train (70%): the model learns from these.
- Validation (15%): we watch these while training, to know when to stop.
- Test (15%): locked away until the very end. One honest score.
from sklearn.model_selection import train_test_split
train, rest = train_test_split(photos, test_size=0.30, stratify=photos["label"], random_state=42)
val, test = train_test_split(rest, test_size=0.50, stratify=rest["label"], random_state=42)
print(len(train), "train,", len(val), "validation,", len(test), "test")
3326 train, 713 validation, 713 test
4. Transfer learning: borrow a network that can already see¶
MobileNetV2 is a deep convolutional neural network, made by Google to run on phones. It was trained on ImageNet: more than a million photos of 1,000 kinds of things. Its early layers learned edges, its middle layers textures and shapes, its last layers whole objects.
We cut off its last layer (the one that says "golden retriever") and keep the rest frozen. Each photo goes in; 1,280 numbers come out that describe what the network sees. Those numbers are the features.
The network wants photos 224 by 224 pixels, with pixel values scaled from -1 to 1.
os.environ["TF_CPP_MIN_LOG_LEVEL"] = "2"
import keras
keras.utils.set_random_seed(42)
SIZE = 224
base = keras.applications.MobileNetV2(include_top=False, pooling="avg", weights="imagenet", input_shape=(SIZE, SIZE, 3))
base.trainable = False
print(f"MobileNetV2: {base.count_params():,} numbers (weights), all frozen. Output: {base.output_shape[1]} features per photo")
def pixels(path):
return np.asarray(Image.open(path).convert("RGB").resize((SIZE, SIZE), Image.BILINEAR), dtype=np.uint8)
def features(paths, batch=64):
out = []
for start in range(0, len(paths), batch):
block = np.stack([pixels(path) for path in paths[start:start + batch]]).astype("float32")
out.append(base.predict(keras.applications.mobilenet_v2.preprocess_input(block), verbose=0))
return np.concatenate(out)
X_train, X_val, X_test = (features(list(part.path)) for part in (train, val, test))
print(X_train.shape, X_val.shape, X_test.shape)
MobileNetV2: 2,257,984 numbers (weights), all frozen. Output: 1280 features per photo
(3326, 1280) (713, 1280) (713, 1280)
5. A baseline first¶
Before any training: what score do you get by always guessing the biggest class? Any real model must beat this.
from sklearn.metrics import accuracy_score, classification_report, confusion_matrix
index = {name: i for i, name in enumerate(CLASSES)}
y_train, y_val, y_test = (part["label"].map(index).to_numpy() for part in (train, val, test))
biggest = np.bincount(y_train).argmax()
baseline = accuracy_score(y_test, np.full_like(y_test, biggest))
print(f"Always guess '{CLASSES[biggest]}': {baseline:.0%} accurate on the test photos")
Always guess 'Plastic': 19% accurate on the test photos
6. Train one new layer¶
One dense layer with nine outputs, one per class, and softmax so the nine scores add up to 100%. It has 1,280 x 9 weights plus 9 biases: about 11,500 numbers to learn, against 2.2 million frozen ones we borrowed.
We stop when the validation score stops improving (early stopping), so the layer does not learn the training photos by heart.
head = keras.Sequential([
keras.Input(shape=(X_train.shape[1],)),
keras.layers.Dropout(0.2),
keras.layers.Dense(len(CLASSES), activation="softmax", kernel_regularizer=keras.regularizers.l2(1e-4)),
])
head.compile(optimizer=keras.optimizers.Adam(1e-3), loss="sparse_categorical_crossentropy", metrics=["accuracy"])
stop = keras.callbacks.EarlyStopping(monitor="val_accuracy", patience=8, restore_best_weights=True)
history = head.fit(X_train, y_train, validation_data=(X_val, y_val), epochs=80, batch_size=32, callbacks=[stop], verbose=0)
print(f"Stopped after {len(history.history['loss'])} rounds (epochs)")
plt.figure(figsize=(6, 3))
plt.plot(history.history["accuracy"], label="training photos")
plt.plot(history.history["val_accuracy"], label="validation photos")
plt.xlabel("epoch"); plt.ylabel("accuracy"); plt.legend(); plt.grid(alpha=0.3)
plt.title("Learning curve")
plt.show()
Stopped after 17 rounds (epochs)
7. Test: one honest score¶
Now, and only now, we open the test photos. Accuracy overall, then per class: precision (when it says glass, how often is it glass?) and recall (of all the glass, how much did it find?).
p_test = head.predict(X_test, verbose=0)
predicted = p_test.argmax(axis=1)
accuracy = accuracy_score(y_test, predicted)
print(f"Test accuracy: {accuracy:.1%} (baseline {baseline:.0%})")
print(classification_report(y_test, predicted, target_names=CLASSES, digits=2))
Test accuracy: 79.4% (baseline 19%)
precision recall f1-score support
Cardboard 0.80 0.87 0.83 69
Food Organics 0.86 0.81 0.83 62
Glass 0.68 0.83 0.74 63
Metal 0.85 0.75 0.80 118
Miscellaneous Trash 0.63 0.68 0.65 75
Paper 0.82 0.85 0.84 75
Plastic 0.77 0.77 0.77 138
Textile Trash 0.87 0.69 0.77 48
Vegetation 0.95 0.94 0.95 65
accuracy 0.79 713
macro avg 0.80 0.80 0.80 713
weighted avg 0.80 0.79 0.79 713
matrix = confusion_matrix(y_test, predicted)
fig, axis = plt.subplots(figsize=(6.5, 5.5))
axis.imshow(matrix, cmap="Greens")
axis.set_xticks(range(len(CLASSES)), CLASSES, rotation=60, ha="right", fontsize=8)
axis.set_yticks(range(len(CLASSES)), CLASSES, fontsize=8)
for i in range(len(CLASSES)):
for j in range(len(CLASSES)):
axis.text(j, i, matrix[i, j], ha="center", va="center", fontsize=8, color="black" if matrix[i, j] < matrix.max() / 2 else "white")
axis.set_xlabel("the model said"); axis.set_ylabel("it really was")
axis.set_title("Confusion matrix (test photos)")
plt.tight_layout()
plt.show()
8. The decision that matters: which bin?¶
Nobody throws away a "class". They choose a bin. In Singapore the blue recycling bin takes paper, cardboard, plastic, glass and metal, if clean and empty. The rest goes to general waste.
Two different mistakes:
- Contamination: general waste put in the blue bin. It can spoil a whole bin of recycling.
- A missed recyclable: something recyclable thrown away. A lost chance, but it spoils nothing else.
BLUE_BIN = {"Cardboard", "Glass", "Metal", "Paper", "Plastic"}
blue = np.array([name in BLUE_BIN for name in CLASSES])
true_blue, said_blue = blue[y_test], blue[predicted]
bin_accuracy = (true_blue == said_blue).mean()
contamination = int((~true_blue & said_blue).sum())
missed_recyclable = int((true_blue & ~said_blue).sum())
print(f"Right bin: {bin_accuracy:.1%} of test photos")
print(f"Contamination (general waste sent to the blue bin): {contamination}")
print(f"Missed recyclables (blue-bin items sent to general waste): {missed_recyclable}")
Right bin: 94.5% of test photos Contamination (general waste sent to the blue bin): 24 Missed recyclables (blue-bin items sent to general waste): 15
9. Knowing when it does not know¶
Softmax always gives some answer, even for a photo of a cat. So we add a rule in code: if the top score is below a threshold, say "not sure" and ask the person to check. Watch the trade-off: a higher threshold answers fewer photos, but gets more of them right.
top = p_test.max(axis=1)
sweep = []
for threshold in (0.0, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9):
sure = top >= threshold
sweep.append({"threshold": threshold, "answered": round(float(sure.mean()), 3),
"accuracy when answered": round(float((predicted[sure] == y_test[sure]).mean()), 3) if sure.any() else None})
pd.DataFrame(sweep)
| threshold | answered | accuracy when answered | |
|---|---|---|---|
| 0 | 0.0 | 1.000 | 0.794 |
| 1 | 0.3 | 0.992 | 0.801 |
| 2 | 0.4 | 0.959 | 0.816 |
| 3 | 0.5 | 0.903 | 0.834 |
| 4 | 0.6 | 0.819 | 0.865 |
| 5 | 0.7 | 0.703 | 0.896 |
| 6 | 0.8 | 0.590 | 0.931 |
| 7 | 0.9 | 0.431 | 0.967 |
UNSURE_BELOW = 0.5
sure = top >= UNSURE_BELOW
print(f"With 'not sure' below {UNSURE_BELOW:.0%}: answers {sure.mean():.0%} of photos, {(predicted[sure] == y_test[sure]).mean():.1%} of those right")
With 'not sure' below 50%: answers 90% of photos, 83.4% of those right
10. Found wrong: look at the mistakes¶
The most useful thing a model does is fail where you can see it. Below are the test photos it got wrong while feeling most sure.
wrong = np.where(predicted != y_test)[0]
worst = wrong[np.argsort(-top[wrong])][:8]
fig, axes = plt.subplots(1, len(worst), figsize=(18, 2.8))
for axis, i in zip(np.atleast_1d(axes), worst):
axis.imshow(Image.open(test.path.iloc[i]))
axis.set_title(f"was {CLASSES[y_test[i]]}\nsaid {CLASSES[predicted[i]]} {top[i]:.0%}", fontsize=8)
axis.axis("off")
plt.show()
Where it will fail next. Every photo here was taken at one landfill in Australia: dirty, crushed, on a similar background. A clean bottle on your desk in Singapore looks different. That is data drift, and the fix is the same as in the scam lab: collect new examples from where the model is used, retrain, re-measure.
11. Responsible AI¶
- Privacy by design. In the web demo the photo is processed on your phone. It is never uploaded, so it cannot leak, be stored or be used for training.
- Advice, not control. The model suggests a bin; a person decides. A wrong suggestion should never lock or charge anyone.
- Say "not sure". Low-confidence answers are shown as "not sure", not as a guess.
- Be transparent. A model card says what the model was trained on, how it was measured and where it fails.
- Credit the data. RealWaste is CC BY 4.0: you may reuse it, if you credit the authors.
12. Assessment: three short tasks¶
- Threshold. Change
UNSURE_BELOWto 0.7. Report the share answered and the accuracy when answered. In two sentences: is 0.7 better for a recycling kiosk? Why? - Contamination first. Which single class pair causes the most contamination (general waste sent to the blue bin)? Suggest one change to the data or the rule that would reduce it.
- Model card. One paragraph: the data and its licence, one number you measured, one place it fails, and one risk if it were used at a real recycling point.
Marking guide (teacher version only)¶
- Threshold (4 marks). 1 mark each for the two figures. 1 mark for stating that answered falls and accuracy rises. 1 mark for a justified choice tied to the cost of contamination.
- Contamination (4 marks). 1 mark for reading the pair from the confusion matrix and the bin mapping. 1 mark for the count. 2 marks for a sensible fix (more photos of that class, a stricter threshold for blue-bin answers only, or asking the person when the two classes are close).
- Model card (4 marks). 1 mark each: data source and licence; a correct figure from their own run; a specific failure (for example, landfill photos versus a clean item at home); a real-world risk (for example, contamination at a recycling point).
13. Fix it for everyone: version 2 learns e-waste¶
Try the web demo with a phone, a charger or a battery. Version 1 has never seen e-waste (RealWaste has none), so it calls a phone metal and sends it to the blue bin. That is the worst mistake it can make: a crushed battery can start a fire in a recycling truck. In Singapore e-waste goes to an e-waste recycling bin, never the blue bin.
"Teach it" in the demo fixes this for one phone. To fix it for everyone, a data team does four things:
- Collect data. Photos of e-waste, with a licence that allows it. We use the E-Tadweer Image Dataset (ETID): Sayed, Hossen, Jano, Hamila, Bensaali, Touati and Varlamis, Mendeley Data, DOI 10.17632/bndsgcspbx.2, licence CC BY 4.0. Four hundred photos of phones, chargers, batteries and cables were picked from it and cut to squares.
- Retrain. A tenth class,
E-waste, and a new layer trained with the same recipe. - Re-measure. On the SAME test photos as version 1, on new e-waste test photos, and on photos taken where the model will be used.
- Release, or not. Only if every check passes. Version 1 is kept, so we can compare, and go back if we must.
This section shows a first attempt that failed, why, and the fix. That is how it usually goes.
EWASTE_URL = "https://d2nke3wqqm57v5.cloudfront.net/clf/vision/data/ewaste.zip"
# On the lab machine the picked photos are already here; in Colab they are downloaded from the demo site (a few MB).
LOCAL_EWASTE = "../ml/data/raw/ewaste"
COLAB_EWASTE = "/content/ewaste"
def photo_folder(local, colab, url):
if os.path.isdir(local):
return local
if not os.path.isdir(colab):
zipfile.ZipFile(io.BytesIO(urllib.request.urlopen(url).read())).extractall(colab)
return colab
EWASTE_ROOT = photo_folder(LOCAL_EWASTE, COLAB_EWASTE, EWASTE_URL)
picked = pd.DataFrame(json.load(open(os.path.join(EWASTE_ROOT, "manifest.json")))["photos"])
picked["path"] = [os.path.join(EWASTE_ROOT, name) for name in picked.file]
print(len(picked), "e-waste photos:", ", ".join(f"{kind} {n}" for kind, n in picked.kind.value_counts().items()))
fig, axes = plt.subplots(1, 8, figsize=(18, 2.4))
for axis, (_, row) in zip(axes, picked.groupby("kind").head(2).iterrows()):
axis.imshow(Image.open(row.path)); axis.set_title(row.kind, fontsize=9); axis.axis("off")
plt.show()
400 e-waste photos: phone 100, charger 100, battery 100, cable 100
# The nine old classes keep version 1's split exactly, so every version is tested on the SAME photos.
# The e-waste photos are split the same way (70 / 15 / 15), with the same mix of kinds in each part.
e_train, e_rest = train_test_split(picked, test_size=0.30, stratify=picked["kind"], random_state=42)
e_val, e_test = train_test_split(e_rest, test_size=0.50, stratify=e_rest["kind"], random_state=42)
CLASSES_V2 = CLASSES + ["E-waste"]
EWASTE = len(CLASSES) # the new class's number
Xe_train, Xe_val, Xe_test = (features(list(part.path)) for part in (e_train, e_val, e_test))
ye_train, ye_val, ye_test = (np.full(len(part), EWASTE) for part in (e_train, e_val, e_test))
def train_head(X, y, X_check, y_check):
"""The same recipe as version 1: one dense softmax layer on the frozen MobileNetV2 features."""
keras.utils.set_random_seed(42)
model = keras.Sequential([
keras.Input(shape=(X.shape[1],)),
keras.layers.Dropout(0.2),
keras.layers.Dense(len(CLASSES_V2), activation="softmax", kernel_regularizer=keras.regularizers.l2(1e-4)),
])
model.compile(optimizer=keras.optimizers.Adam(1e-3), loss="sparse_categorical_crossentropy", metrics=["accuracy"])
stop = keras.callbacks.EarlyStopping(monitor="val_accuracy", patience=8, restore_best_weights=True)
fitted = model.fit(X, y, validation_data=(X_check, y_check), epochs=80, batch_size=32, callbacks=[stop], verbose=0)
return model, fitted
# First attempt: the old photos plus the e-waste photos.
head_try1, history_try1 = train_head(np.concatenate([X_train, Xe_train]), np.concatenate([y_train, ye_train]),
np.concatenate([X_val, Xe_val]), np.concatenate([y_val, ye_val]))
print(f"First attempt: {len(CLASSES_V2)} classes, stopped after {len(history_try1.history['loss'])} epochs")
First attempt: 10 classes, stopped after 25 epochs
Measure the first attempt: the lab test¶
- Regression check. On the old test photos (no e-waste), is it as good as version 1? A new version that breaks what worked is not released.
- The new class. Of the e-waste test photos, how many does it find?
- The mistake that matters. How many e-waste test photos would each version send to the blue bin?
blue_v2 = np.append(blue, False) # e-waste never goes in the blue bin
OLD = len(y_test)
old_v1 = accuracy_score(y_test, predicted)
said_v1_e = head.predict(Xe_test, verbose=0).argmax(axis=1) # version 1 on the e-waste test photos
def lab_test(model):
old = accuracy_score(y_test, model.predict(X_test, verbose=0).argmax(axis=1))
said_e = model.predict(Xe_test, verbose=0).argmax(axis=1)
return {"old": old, "found": float((said_e == EWASTE).mean()), "blue": int(blue_v2[said_e].sum())}
try1 = lab_test(head_try1)
pd.DataFrame({"version 1": [f"{old_v1:.1%}", "0% (no e-waste class)", f"{blue[said_v1_e].sum()} of {len(e_test)}"],
"first attempt": [f"{try1['old']:.1%}", f"{try1['found']:.0%}", f"{try1['blue']} of {len(e_test)}"]},
index=["accuracy on the old test photos", "e-waste test photos found", "e-waste sent to the blue bin"])
| version 1 | first attempt | |
|---|---|---|
| accuracy on the old test photos | 79.4% | 82.2% |
| e-waste test photos found | 0% (no e-waste class) | 98% |
| e-waste sent to the blue bin | 42 of 60 | 1 of 60 |
It looks excellent. Before releasing it, one more test.
The honest test: photos taken in Singapore¶
Every test so far used photos from the same collections the model learned from. The real question is how it does on photos from where it will be used. Below are 42 photos taken on a phone in Singapore (food, plants, shoes, a few household items and three pieces of e-waste), labelled by hand and never used for training. Photos with people, faces, number plates, names or screens of personal data were left out.
We count three things for each version: the right material, the right bin (blue, e-waste, or neither), and the mistake that spoils recycling: something sent to the blue bin that does not belong there.
OWN_URL = "https://d2nke3wqqm57v5.cloudfront.net/clf/vision/samples/"
LOCAL_SAMPLES = "../web/clf/vision/samples"
def own_photos():
if os.path.isdir(LOCAL_SAMPLES):
index = json.load(open(os.path.join(LOCAL_SAMPLES, "own.json")))
return [(os.path.join(LOCAL_SAMPLES, s["file"]), s["classId"]) for s in index["samples"]]
index = json.load(urllib.request.urlopen(OWN_URL + "own.json")) # in Colab: from the demo site
rows = []
for s in index["samples"]:
local = os.path.join("/content", s["file"])
os.makedirs(os.path.dirname(local), exist_ok=True)
urllib.request.urlretrieve(OWN_URL + s["file"], local)
rows.append((local, s["classId"]))
return rows
own = pd.DataFrame(own_photos(), columns=["path", "classId"])
X_own = features(list(own.path))
IDS_V2 = [name.lower().replace(" ", "_") for name in CLASSES] + ["ewaste"]
BLUE_IDS = {IDS_V2[i] for i in np.where(blue_v2)[0]}
bin_of = lambda class_id: "blue" if class_id in BLUE_IDS else "e-waste" if class_id == "ewaste" else "neither"
def real_world(model):
said = [IDS_V2[i] for i in model.predict(X_own, verbose=0).argmax(axis=1)]
truth = list(own.classId)
return {"right": int(sum(s == t for s, t in zip(said, truth))),
"rightBin": int(sum(bin_of(s) == bin_of(t) for s, t in zip(said, truth))),
"wrongBlue": int(sum(bin_of(s) == "blue" and bin_of(t) != "blue" for s, t in zip(said, truth))),
"saidEwaste": int(sum(s == "ewaste" for s in said))}
own_v1, own_try1 = real_world(head), real_world(head_try1)
ROWS = ["right material", "right bin", "sent to the blue bin wrongly", "called e-waste"]
KEYS = ["right", "rightBin", "wrongBlue", "saidEwaste"]
print(f"{len(own)} photos taken in Singapore:", own.classId.value_counts().to_dict())
pd.DataFrame({"version 1": [f"{own_v1[k]} of {len(own)}" for k in KEYS], "first attempt": [f"{own_try1[k]} of {len(own)}" for k in KEYS]}, index=ROWS)
42 photos taken in Singapore: {'food_organics': 19, 'vegetation': 11, 'textile_trash': 4, 'ewaste': 3, 'metal': 2, 'cardboard': 2, 'plastic': 1}
| version 1 | first attempt | |
|---|---|---|
| right material | 14 of 42 | 9 of 42 |
| right bin | 26 of 42 | 14 of 42 |
| sent to the blue bin wrongly | 14 of 42 | 0 of 42 |
| called e-waste | 0 of 42 | 30 of 42 |
What went wrong: a shortcut¶
The first attempt calls flowers, cake and coffee e-waste, often at more than 90%. Why? Look at where each class came from:
- Every e-waste photo is a clean, well-lit phone photo of one object on a table.
- Every other photo is a dirty landfill photo on the same grey background.
The easiest rule that fits the training data is not "this is a phone" but "a clean photo is e-waste". The model learned the shortcut, the photo style, instead of the object. The lab test could not catch it, because its e-waste test photos came from the same collection. Only photos from somewhere new could.
The fix: break the shortcut with data. Add clean, everyday photos of the OTHER materials (food, flowers, shoes, bottles, cans, boxes, books), so a clean photo no longer means e-waste. We use Open Images V7 (Google): photos listed as CC BY 2.0, each credited to its author; photos showing people were left out.
EVERYDAY_URL = "https://d2nke3wqqm57v5.cloudfront.net/clf/vision/data/everyday.zip"
EVERYDAY_ROOT = photo_folder("../ml/data/raw/everyday", "/content/everyday", EVERYDAY_URL)
everyday = pd.DataFrame(json.load(open(os.path.join(EVERYDAY_ROOT, "manifest.json")))["photos"])
everyday["path"] = [os.path.join(EVERYDAY_ROOT, name) for name in everyday.file]
everyday["y"] = everyday.classId.map({class_id: i for i, class_id in enumerate(IDS_V2)})
print(len(everyday), "everyday photos:", everyday.classId.value_counts().to_dict())
d_train, d_rest = train_test_split(everyday, test_size=0.30, stratify=everyday["classId"], random_state=42)
d_val, d_test = train_test_split(d_rest, test_size=0.50, stratify=d_rest["classId"], random_state=42)
Xd_train, Xd_val, Xd_test = (features(list(part.path)) for part in (d_train, d_val, d_test))
fig, axes = plt.subplots(1, 8, figsize=(18, 2.4))
for axis, (_, row) in zip(axes, everyday.groupby("classId").head(1).iterrows()):
axis.imshow(Image.open(row.path)); axis.set_title(row.classId, fontsize=9); axis.axis("off")
plt.show()
# Version 2: the old photos, the e-waste photos and the everyday photos. Same recipe again.
head_v2, history_v2 = train_head(np.concatenate([X_train, Xe_train, Xd_train]), np.concatenate([y_train, ye_train, d_train.y.to_numpy()]),
np.concatenate([X_val, Xe_val, Xd_val]), np.concatenate([y_val, ye_val, d_val.y.to_numpy()]))
print(f"Version 2 stopped after {len(history_v2.history['loss'])} epochs")
292 everyday photos: {'food_organics': 50, 'vegetation': 50, 'textile_trash': 50, 'glass': 50, 'metal': 48, 'cardboard': 25, 'paper': 14, 'plastic': 5}
Version 2 stopped after 17 epochs
Re-measure version 2: three release checks¶
- Regression check. No worse than version 1 on the old test photos (within two photos: noise at this size).
- The new class. Far fewer e-waste test photos sent to the blue bin than version 1.
- The real world. On the Singapore photos: at least as many right bins as version 1, and no more things sent to the blue bin wrongly.
X2_test = np.concatenate([X_test, Xe_test, Xd_test])
y2_test = np.concatenate([y_test, ye_test, d_test.y.to_numpy()])
p2_test = head_v2.predict(X2_test, verbose=0)
predicted_v2 = p2_test.argmax(axis=1)
E_END = OLD + len(e_test) # test photos: version 1's own, then e-waste, then everyday
v2 = lab_test(head_v2)
old_v2 = v2["old"]
said_v2_e = predicted_v2[OLD:E_END]
own_v2 = real_world(head_v2)
everyday_v1 = float((head.predict(Xd_test, verbose=0).argmax(axis=1) == d_test.y.to_numpy()).mean())
everyday_v2 = float((predicted_v2[E_END:] == d_test.y.to_numpy()).mean())
pd.DataFrame({
"version 1": [f"{old_v1:.1%}", "0%", f"{blue[said_v1_e].sum()} of {len(e_test)}", f"{everyday_v1:.0%}"] + [f"{own_v1[k]} of {len(own)}" for k in KEYS],
"first attempt": [f"{try1['old']:.1%}", f"{try1['found']:.0%}", f"{try1['blue']} of {len(e_test)}", "not tested"] + [f"{own_try1[k]} of {len(own)}" for k in KEYS],
"version 2": [f"{old_v2:.1%}", f"{v2['found']:.0%}", f"{v2['blue']} of {len(e_test)}", f"{everyday_v2:.0%}"] + [f"{own_v2[k]} of {len(own)}" for k in KEYS],
}, index=["old test photos right", "e-waste test photos found", "e-waste sent to the blue bin", "everyday test photos right"] + [f"Singapore: {r}" for r in ROWS])
| version 1 | first attempt | version 2 | |
|---|---|---|---|
| old test photos right | 79.4% | 82.2% | 80.5% |
| e-waste test photos found | 0% | 98% | 97% |
| e-waste sent to the blue bin | 42 of 60 | 1 of 60 | 2 of 60 |
| everyday test photos right | 64% | not tested | 89% |
| Singapore: right material | 14 of 42 | 9 of 42 | 31 of 42 |
| Singapore: right bin | 26 of 42 | 14 of 42 | 34 of 42 |
| Singapore: sent to the blue bin wrongly | 14 of 42 | 0 of 42 | 5 of 42 |
| Singapore: called e-waste | 0 of 42 | 30 of 42 | 5 of 42 |
# Regression check, new class, real world: the three release checks, in code.
checks = {
"no worse on the old test photos": old_v2 >= old_v1 - 2 / OLD,
"far fewer e-waste photos to the blue bin": v2["blue"] < blue[said_v1_e].sum() / 2,
"Singapore: at least as many right bins": own_v2["rightBin"] >= own_v1["rightBin"],
"Singapore: no more wrong blue-bin answers": own_v2["wrongBlue"] <= own_v1["wrongBlue"],
}
for name, passed in checks.items():
print(("PASS " if passed else "FAIL ") + name)
first_attempt_passes = (try1["old"] >= old_v1 - 2 / OLD and own_try1["rightBin"] >= own_v1["rightBin"])
print("Release version 2:", all(checks.values()), "| the first attempt would have passed:", first_attempt_passes)
false_ewaste = int((predicted_v2[:OLD] == EWASTE).sum())
print(f"Old test photos version 2 wrongly calls e-waste: {false_ewaste} of {OLD}")
PASS no worse on the old test photos PASS far fewer e-waste photos to the blue bin PASS Singapore: at least as many right bins PASS Singapore: no more wrong blue-bin answers Release version 2: True | the first attempt would have passed: False Old test photos version 2 wrongly calls e-waste: 0 of 713
Release decision. A version is released only when every check passes. The first attempt passed the lab test and failed the real world; version 2 had to pass both. The web demo shows both released versions side by side on the same photo, and the model card shows the figures, including the failed attempt.
Where it will still fail. Forty-two Singapore photos are a small test, and they are mostly food and plants. A wider local test set, collected the same careful way, is the next step: collect, retrain, re-measure, release.
Assessment task 4 (release). From the tables above: why did the first attempt look good and still fail? Which single check caught it? Would you release version 2? Give the figures your decision rests on, and one risk that remains.
Marking guide, task 4 (teacher version only)¶
Release (6 marks). 2 marks for explaining the shortcut (photo style, not the object, because each class came from one collection). 1 mark for naming the Singapore check. 1 mark for reading version 2's figures against version 1's. 1 mark for a decision that follows from the checks. 1 mark for a remaining risk (a small local test set; new e-waste kinds such as vapes or power banks; false e-waste on metal cans).
Appendix: export the model for the web demo¶
This runs only on the lab machine, where the demo's folders exist. It writes the trained layer as plain JSON, every measured figure, one sample photo per class from the test set, and the parity cases the demo's tests use to prove the browser gives exactly these probabilities. In Colab it does nothing.
WEB_DIR = "../web/clf/vision"
PARITY_DIR = "../vision/parity"
CLASS_IDS = [name.lower().replace(" ", "_") for name in CLASSES]
if os.path.isdir(WEB_DIR) and os.path.isdir(PARITY_DIR):
weights, bias = head.layers[-1].get_weights()
model_out = {
"format": "mobilenetv2-softmax-head/1",
"featureModel": "MobileNetV2 1.0, 224 x 224, ImageNet weights, feature vector (1,280 numbers)",
"inputSize": SIZE,
"classes": CLASS_IDS,
"classNames": CLASSES,
"features": int(weights.shape[0]),
"weights": np.round(weights, 7).tolist(),
"bias": np.round(bias, 7).tolist(),
"unsureBelow": UNSURE_BELOW,
"kerasVersion": keras.__version__,
}
report = classification_report(y_test, predicted, target_names=CLASSES, output_dict=True, zero_division=0)
metrics = {
"data": {"photos": int(len(photos)), "train": int(len(train)), "validation": int(len(val)), "test": int(len(test)),
"perClass": {CLASS_IDS[i]: int(counts[name]) for i, name in enumerate(CLASSES)}},
"baseline": {"accuracy": round(float(baseline), 4), "alwaysSays": CLASS_IDS[biggest]},
"epochs": int(len(history.history["loss"])),
"trainAccuracy": round(float(max(history.history["accuracy"])), 4),
"validationAccuracy": round(float(max(history.history["val_accuracy"])), 4),
"test": {"accuracy": round(float(accuracy), 4), "photos": int(len(test)), "right": int((predicted == y_test).sum())},
"perClass": {CLASS_IDS[i]: {"precision": round(report[name]["precision"], 4), "recall": round(report[name]["recall"], 4),
"photos": int(report[name]["support"])} for i, name in enumerate(CLASSES)},
"confusion": matrix.tolist(),
"bin": {"accuracy": round(float(bin_accuracy), 4), "contamination": contamination, "missedRecyclable": missed_recyclable,
"blueBin": [CLASS_IDS[i] for i, name in enumerate(CLASSES) if name in BLUE_BIN]},
"unsureBelow": UNSURE_BELOW,
"unsureSweep": sweep,
"answeredAtThreshold": round(float(sure.mean()), 4),
"accuracyWhenAnswered": round(float((predicted[sure] == y_test[sure]).mean()), 4),
}
os.makedirs(os.path.join(WEB_DIR, "model"), exist_ok=True)
os.makedirs(os.path.join(WEB_DIR, "samples"), exist_ok=True)
samples, cases = [], []
for i, name in enumerate(CLASSES):
row = np.where(y_test == i)[0][0] # the FIRST test photo of each class, right or wrong
file = f"{CLASS_IDS[i]}.jpg"
Image.fromarray(pixels(test.path.iloc[row])).save(os.path.join(WEB_DIR, "samples", file), quality=90)
samples.append({"file": file, "classId": CLASS_IDS[i], "source": os.path.basename(test.path.iloc[row])})
for i in range(len(CLASSES)):
row = np.where(y_test == i)[0][1] # parity: a second test photo per class, as exact pixels
name = f"case-{CLASS_IDS[i]}.rgb"
pixels(test.path.iloc[row]).tofile(os.path.join(PARITY_DIR, name))
cases.append({"file": name, "classId": CLASS_IDS[i], "probabilities": [round(float(v), 7) for v in p_test[row]]})
metrics["samples"] = samples
for path, payload in ((os.path.join(WEB_DIR, "model", "head.json"), model_out), (os.path.join(WEB_DIR, "model", "metrics.json"), metrics),
(os.path.join(PARITY_DIR, "parity.json"), {"note": "Probabilities from the notebook for exact 224 x 224 pixels; RealWaste test photos (Single, Iranmanesh and Raad 2023, CC BY 4.0, DOI 10.24432/C5SS4G).", "cases": cases})):
with open(path, "w", encoding="utf-8") as handle:
json.dump(payload, handle, separators=(",", ":"))
print(f"Exported: head.json, metrics.json, {len(samples)} sample photos, {len(cases)} parity cases")
else:
print("Not on the lab machine: nothing exported.")
Exported: head.json, metrics.json, 9 sample photos, 9 parity cases
# Version 2's export: its own folder, so version 1's files are never overwritten.
V2_DIR = os.path.join(WEB_DIR, "model", "v2")
CLASS_IDS_V2 = CLASS_IDS + ["ewaste"]
EWASTE_SAMPLES = 6
def features_of_pixels(px):
return base.predict(keras.applications.mobilenet_v2.preprocess_input(px[None].astype("float32")), verbose=0)
if os.path.isdir(WEB_DIR) and os.path.isdir(PARITY_DIR):
weights2, bias2 = head_v2.layers[-1].get_weights()
head2_out = {**model_out, "classes": CLASS_IDS_V2, "classNames": CLASSES_V2, "features": int(weights2.shape[0]),
"weights": np.round(weights2, 7).tolist(), "bias": np.round(bias2, 7).tolist(), "version": "v2"}
report2 = classification_report(y2_test, predicted_v2, target_names=CLASSES_V2, output_dict=True, zero_division=0)
sure2 = p2_test.max(axis=1) >= UNSURE_BELOW
true_blue2, said_blue2 = blue_v2[y2_test], blue_v2[predicted_v2]
metrics2 = {
"version": "v2",
"data": {"photos": int(len(photos) + len(picked) + len(everyday)), "train": int(len(y_train) + len(e_train) + len(d_train)),
"validation": int(len(y_val) + len(e_val) + len(d_val)), "test": int(len(y2_test)),
"everyday": {"photos": int(len(everyday)), "perClass": {k: int(n) for k, n in everyday.classId.value_counts().items()},
"train": int(len(d_train)), "validation": int(len(d_val)), "test": int(len(d_test))},
"ewaste": {"photos": int(len(picked)), "kinds": {k: int(n) for k, n in picked.kind.value_counts().items()},
"train": int(len(e_train)), "validation": int(len(e_val)), "test": int(len(e_test))}},
"epochs": int(len(history_v2.history["loss"])),
"test": {"accuracy": round(float(accuracy_score(y2_test, predicted_v2)), 4), "photos": int(len(y2_test)), "right": int((predicted_v2 == y2_test).sum())},
"oldTest": {"photos": int(OLD), "v1": round(float(old_v1), 4), "v2": round(float(old_v2), 4)},
"ewasteTest": {"photos": int(len(e_test)), "found": round(float((said_v2_e == EWASTE).mean()), 4),
"blueBinV1": int(blue[said_v1_e].sum()), "blueBinV2": int(blue_v2[said_v2_e].sum()),
"v1Said": {CLASS_IDS[i]: int((said_v1_e == i).sum()) for i in range(len(CLASSES)) if (said_v1_e == i).any()},
"falseEwaste": false_ewaste},
"perClass": {CLASS_IDS_V2[i]: {"precision": round(report2[name]["precision"], 4), "recall": round(report2[name]["recall"], 4),
"photos": int(report2[name]["support"])} for i, name in enumerate(CLASSES_V2)},
"confusion": confusion_matrix(y2_test, predicted_v2).tolist(),
"bin": {"accuracy": round(float((true_blue2 == said_blue2).mean()), 4), "contamination": int((~true_blue2 & said_blue2).sum()),
"missedRecyclable": int((true_blue2 & ~said_blue2).sum()), "blueBin": [CLASS_IDS_V2[i] for i in np.where(blue_v2)[0]]},
"unsureBelow": UNSURE_BELOW,
"answeredAtThreshold": round(float(sure2.mean()), 4),
"accuracyWhenAnswered": round(float((predicted_v2[sure2] == y2_test[sure2]).mean()), 4),
}
os.makedirs(V2_DIR, exist_ok=True)
samples2, cases2 = [], []
# Six e-waste test photos for the page (36 + 6 fills rows of 3 and 6), taking the kinds in turn: phone, charger, ...
order = {kind: i for i, kind in enumerate(picked.kind.unique())}
turns = e_test.assign(turn=e_test.groupby("kind").cumcount(), rank=e_test.kind.map(order)).sort_values(["turn", "rank"])
for n, (_, row) in enumerate(turns.head(EWASTE_SAMPLES).iterrows(), start=1):
file = f"ewaste-{n}.jpg"
Image.fromarray(pixels(row.path)).save(os.path.join(WEB_DIR, "samples", file), quality=90)
samples2.append({"file": file, "classId": "ewaste", "kind": row.kind, "source": row.source})
for case in json.load(open(os.path.join(PARITY_DIR, "parity.json")))["cases"]: # version 1's parity photos, through version 2
x = features_of_pixels(np.fromfile(os.path.join(PARITY_DIR, case["file"]), dtype=np.uint8).reshape(SIZE, SIZE, 3))
cases2.append({"file": case["file"], "classId": case["classId"], "probabilities": [round(float(v), 7) for v in head_v2.predict(x, verbose=0)[0]]})
for n in range(2): # and two e-waste test photos, as exact pixels
name = f"case-ewaste-{n + 1}.rgb"
px = pixels(e_test.path.iloc[n]); px.tofile(os.path.join(PARITY_DIR, name))
cases2.append({"file": name, "classId": "ewaste", "probabilities": [round(float(v), 7) for v in head_v2.predict(features_of_pixels(px), verbose=0)[0]]})
metrics2["samples"] = samples2
metrics2["ownTest"] = {"photos": int(len(own)), "kinds": {k: int(n) for k, n in own.classId.value_counts().items()}, "v1": own_v1, "v2": own_v2}
metrics2["everydayTest"] = {"photos": int(len(d_test)), "v1": round(everyday_v1, 4), "v2": round(everyday_v2, 4)}
metrics2["firstAttempt"] = {"oldTest": round(float(try1["old"]), 4), "ewasteFound": round(try1["found"], 4), "blueBin": try1["blue"],
"ownTest": own_try1, "released": False}
metrics2["releaseChecks"] = {name: bool(passed) for name, passed in checks.items()}
for path, payload in ((os.path.join(V2_DIR, "head.json"), head2_out), (os.path.join(V2_DIR, "metrics.json"), metrics2),
(os.path.join(PARITY_DIR, "parity-v2.json"), {"note": "Version 2's probabilities for exact 224 x 224 pixels: version 1's parity photos plus two ETID e-waste test photos (CC BY 4.0, DOI 10.17632/bndsgcspbx.2).", "cases": cases2})):
with open(path, "w", encoding="utf-8") as handle:
json.dump(payload, handle, separators=(",", ":"))
print(f"Exported version 2: head.json, metrics.json, {len(samples2)} e-waste sample photos, {len(cases2)} parity cases")
else:
print("Not on the lab machine: nothing exported.")
Exported version 2: head.json, metrics.json, 6 e-waste sample photos, 11 parity cases