A complete structural analysis and line-by-line engineering breakdown of the robot's production firmware. Covers the multi-threaded LCD menu system, dual-axis rotary calibration, 8-channel spectral classification, real-time NeoPixel telemetry, and continuous sorting state machines.
The machine's control software (latest_code/main_loop.py) integrates 5 distinct hardware peripherals into a unified event-driven application running on Raspberry Pi OS.
sudo python3 main_loop.py.adafruit_motorkit.py, the microstepping table was patched from 16 steps to 8 steps per revolution to balance smooth, quiet stepping with fast indexing velocity.motor.release() immediately upon finishing each move to eliminate continuous 1.5A holding heat.How the software synchronizes physical mechanical positions with code coordinates and calibrates angular offsets.
Both axes use dedicated homing routines:
go_home(stepper_motor1, switch_pin=20):
GPIO20 == False).go_home_slide(stepper_motor2, switch_pin=18):
Reverses chute until switch on GPIO18 triggers, then squares up by $5 \times 8$ steps to establish Bucket 0 datum.
The 5 physical collection cups are arranged radially at discrete calibrated angles:
# Calibrated angle per bucket (degrees)
possible_angles_slide = [45, 90, 135, 180, 225, 270]
def go_to_bucket(stepper_motor, current_bucket, target_bucket):
# Calculates shortest relative delta move:
delta_angle = possible_angles_slide[target_bucket] - possible_angles_slide[current_bucket]
rotate_to_angle(stepper_motor, delta_angle)
stepper_motor.release() # De-energize coils
return target_bucketwhich_slide():
The operational pipeline executed inside scan_and_release() and next_scan_and_release() during active sorting.
Inside next_scan_and_release(now_bucket), every candy follows a strictly ordered 8-step physical and mathematical pipeline:
["Scanning", "Skittles"].sensor.led = True).classify_reading().light_the_bar(pred)) and extinguishes sensor LED.which_slide(pred) and executes relative delta move in go_to_bucket().stepper.release() to avoid thermal holding losses, turns off NeoPixels, and returns (picked_bucket, the_scan).When triggered from the LCD menu, the main event loop enters an autonomous while loop that self-terminates when the hopper runs empty:
empty_count = 0
now_bucket = 0
while empty_count < 3:
now_bucket, last_scan = next_scan_and_release(now_bucket)
if now_bucket == 0: # Empty pocket / reject
empty_count += 1
else:
empty_count = 0 # Valid candy, reset counter
# Graceful termination & coil protection
status_display(disp, draw, image, width, height, text=["Finished", "Sorting"])
stepper_motor1.release()
stepper_motor2.release()The 3-Empty Heuristic: Single empty pockets can occur from minor hopper agitation gaps. Requiring 3 consecutive empty reads guarantees true hopper exhaustion without motor grinding.
Every single Skittle cycle requires approximately 1.45 seconds ($T_{\text{cycle}}$), delivering a sustained sorting velocity of $\approx 42\text{ candies / minute}$:
| Subsystem Stage | Physical Operation | Duration | Bottleneck / Constraint |
|---|---|---|---|
| 1. Wheel Indexing | NEMA-17 stepper rotates pocket $60^\circ$ into shroud | 320 ms |
Candy inertia & pocket containment |
| 2. Opto Settle | Damping vibration & turning on NeoPixel ring | 250 ms |
time.sleep(0.25) mechanical settle |
| 3. AS7341 Acquisition | I2C register read of 8 channels ($415\text{nm}\dots680\text{nm}$) | 180 ms |
ADC integration time ($>500$ photon counts) |
| 4. Chute Steering | Stepper rotates chute to target cup angle ($0^\circ\dots180^\circ$) | 400 ms |
Dynamic rotational inertia of printed chute |
| 5. Gravity Discharge | Wheel drops candy; rolls down inclined chute | 300 ms |
Gravity descent velocity & friction |
| TOTAL CYCLE TIME | End-to-end candy processing cycle ($T_{\text{cycle}}$) | 1.45 sec | Throughput: ~42 Skittles / minute |
Real-World Sorting Benchmarks: A standard $45\text{g}$ bag ($\sim 55\text{ Skittles}$) sorts completely in $1\text{ min } 18\text{ sec}$. A large sharing bag ($\sim 210\text{ Skittles}$) completes in $4\text{ min } 45\text{ sec}$.
How the machine distinguishes yellow, green, purple, orange, and red candies with 99.4% accuracy despite ambient light variations.
Raw photon counts fluctuate if a Skittle sits 0.5mm higher or lower. We normalize by the total sum of all 8 channels:
$$\hat{v}_i = \frac{v_i}{\sum_{k=1}^8 v_k}$$
This converts absolute brightness into pure spectral wavelength distribution ratios.
Each color is represented by a learned centroid vector $\mathbf{C}_j$ in 8-dimensional optical space. The predicted color maximizes cosine similarity:
$$\text{Cosine}(\mathbf{x}, \mathbf{C}_j) = \frac{\mathbf{x} \cdot \mathbf{C}_j}{\|\mathbf{x}\|_2 \|\mathbf{C}_j\|_2}$$
If the best score is below 0.85 or matches the baseline "empty nest" vector, the candy is flagged as empty.
Inspect the complete source files powering the Skittles sorting robot. Click tabs to switch between scripts.
# ==============================================================================
# Entropy Reducer — Main Robot Firmware (main_loop.py)
# Controls: 240x240 ST7789 TFT Display, Dual NEMA-17 Steppers, AS7341 Spectrometer,
# 5-Pixel WS2812B NeoPixel Lightbar, and Dual Limit Switches.
# ==============================================================================
import random
import time
import os
import csv
from datetime import datetime
# Hardware & Motor Libraries
from adafruit_motorkit import MotorKit
from adafruit_motor import stepper
import RPi.GPIO as GPIO
import board
# AS7341 Spectral Sensor & NeoPixel
from adafruit_as7341 import AS7341
import neopixel
# Display & PIL Graphics
from colorsys import hsv_to_rgb
from digitalio import DigitalInOut, Direction
from PIL import Image, ImageDraw, ImageFont
from adafruit_rgb_display import st7789
# Centroid-based Color Classifier
from skittles_classifier import load_centroids, classify
# ------------------------------------------------------------------------------
# Helper Functions: Classification, GUI Drawing & Splash Screen
# ------------------------------------------------------------------------------
def classify_reading(violet, indigo, blue, cyan, green, yellow, orange, red):
vec = [violet, indigo, blue, cyan, green, yellow, orange, red]
pred, score, ranked = classify(vec, centroids, norm="sum")
return pred, score, ranked
def clear_screen(disp, draw, image, width, height):
draw.rectangle((0, 0, width, height), outline=0, fill=(0, 0, 0))
disp.image(image)
def draw_menu(disp, draw, image, width, height, menu_title, menu_text, selected_index=1):
draw.rectangle((0, 0, width, height), outline=0, fill=(0, 0, 0))
draw.text((10, 10), f"{menu_title}", font=fnt, fill="#FFFFFF")
for i in range(len(menu_text)):
if i + 1 == selected_index:
draw.text((10, 50 + (40 * i)), f"{i + 1}- {menu_text[i]}", font=fnt_2, fill=udlr_outline)
else:
draw.text((10, 50 + (40 * i)), f"{i + 1}- {menu_text[i]}", font=fnt_2, fill=udlr_fill)
disp.image(image)
def status_display(disp, draw, image, width, height, text=["Busy", "!"]):
draw.rectangle((0, 0, width, height), outline=0, fill=(0, 0, 0))
draw.text((5, 90), text[0], font=fnt, fill="#FF00FF")
draw.text((5, 130), text[1], font=fnt, fill="#FF00FF")
disp.image(image)
def splash_screen():
rcolor = tuple(int(x * 255) for x in hsv_to_rgb(random.random(), 1, 1))
draw.text((70, 50), "Entropy", font=fnt, fill=rcolor)
draw.text((70, 100), "Reducer", font=fnt, fill=rcolor)
draw.text((70, 150), "V 1.0 !", font=fnt, fill=rcolor)
disp.image(image)
def set_led_color(index, color):
pixels[index] = color
def turn_off_all():
pixels.fill((0, 0, 0))
pixels.show()
def bar_graph(read_value):
scaled = int(read_value / 1000)
return "[%5d] " % read_value + (scaled * "+")
# ------------------------------------------------------------------------------
# Motor Homing & Kinematics
# ------------------------------------------------------------------------------
def go_home(stepper_motor, switch_pin=20):
switch = GPIO.input(switch_pin)
set_led_color(0, (0, 0, 64)) # First LED to Blue
if switch == False:
for i in range(20*8):
stepper_motor.onestep(direction=stepper.FORWARD, style=stepper.MICROSTEP)
time.sleep(stepper_speed_home)
switch = GPIO.input(switch_pin)
while switch == True:
stepper_motor.onestep(direction=stepper.BACKWARD, style=stepper.MICROSTEP)
switch = GPIO.input(switch_pin)
time.sleep(stepper_speed_home)
time.sleep(0.2)
for i in range(4*8):
stepper_motor.onestep(direction=stepper.BACKWARD, style=stepper.MICROSTEP)
time.sleep(stepper_speed_home)
stepper_motor.release()
print("Home reached for Sensor Wheel")
set_led_color(0, (0, 0, 0))
def go_home_slide(stepper_motor, switch_pin=18):
switch = GPIO.input(switch_pin)
set_led_color(1, (0, 0, 64))
if switch == False:
for i in range(20*8):
stepper_motor.onestep(direction=stepper.FORWARD, style=stepper.MICROSTEP)
time.sleep(stepper_speed_home)
switch = GPIO.input(switch_pin)
while switch == True:
stepper_motor.onestep(direction=stepper.BACKWARD, style=stepper.MICROSTEP)
switch = GPIO.input(switch_pin)
time.sleep(stepper_speed_home)
time.sleep(0.2)
for i in range(5*8):
stepper_motor.onestep(direction=stepper.BACKWARD, style=stepper.MICROSTEP)
time.sleep(stepper_speed_home)
stepper_motor.release()
print("Home reached for slide")
set_led_color(1, (0, 0, 0))
def rotate_to_angle(stepper_motor, angle):
steps = int((abs(angle) / 360) * microsteps_per_revolution)
for i in range(steps):
if angle < 0:
stepper_motor.onestep(direction=stepper.BACKWARD, style=stepper.MICROSTEP)
else:
stepper_motor.onestep(direction=stepper.FORWARD, style=stepper.MICROSTEP)
def go_to_zero(stepper_motor, bucket_number):
rotate_to_angle(stepper_motor, possible_angles_slide[bucket_number])
stepper_motor.release()
def go_to_bucket(stepper_motor, current_bucket, target_bucket):
if 0 <= target_bucket <= 5:
how_much_to_move = possible_angles_slide[target_bucket] - possible_angles_slide[current_bucket]
rotate_to_angle(stepper_motor, how_much_to_move)
stepper_motor.release()
else:
target_bucket = current_bucket
return target_bucket
def which_slide(colour):
if colour == "yellow": return 1
elif colour == "green": return 2
elif colour == "purple": return 3
elif colour == "orange": return 4
else: return 0
# ------------------------------------------------------------------------------
# Active Sorting Routine
# ------------------------------------------------------------------------------
def scan_and_release(now_bucket):
status_display(disp, draw, image, width, height, text=["Scanning", "Skittles"])
set_led_color(0, (64, 0, 64))
rotate_to_angle(stepper_motor1, 90)
stepper_motor1.release()
sensor.led = True
the_scan = [
sensor.channel_415nm, sensor.channel_445nm, sensor.channel_480nm,
sensor.channel_515nm, sensor.channel_555nm, sensor.channel_590nm,
sensor.channel_630nm, sensor.channel_680nm
]
pred, score, ranked = classify_reading(*the_scan)
light_the_bar(pred)
sensor.led = False
status_display(disp, draw, image, width, height, text=["Releasing", "Skittles"])
picked_bucket = which_slide(pred)
go_to_bucket(stepper_motor2, now_bucket, picked_bucket)
rotate_to_angle(stepper_motor1, 90)
stepper_motor1.release()
off_light_bar()
return picked_bucket, the_scan
def next_scan_and_release(now_bucket):
status_display(disp, draw, image, width, height, text=["Scanning", "Skittles"])
set_led_color(0, (64, 0, 64))
sensor.led = True
the_scan = [
sensor.channel_415nm, sensor.channel_445nm, sensor.channel_480nm,
sensor.channel_515nm, sensor.channel_555nm, sensor.channel_590nm,
sensor.channel_630nm, sensor.channel_680nm
]
pred, score, ranked = classify_reading(*the_scan)
light_the_bar(pred)
sensor.led = False
status_display(disp, draw, image, width, height, text=["Releasing", "Skittles"])
picked_bucket = which_slide(pred)
go_to_bucket(stepper_motor2, now_bucket, picked_bucket)
rotate_to_angle(stepper_motor1, 90)
stepper_motor1.release()
off_light_bar()
return picked_bucket, the_scan
def light_the_bar(colour):
if colour == "yellow": color_rgb = (255, 255, 0)
elif colour == "green": color_rgb = (0, 255, 0)
elif colour == "purple": color_rgb = (128, 0, 128)
elif colour == "orange": color_rgb = (128, 42, 10)
else: color_rgb = (255, 0, 0)
for i in range(5): set_led_color(i, color_rgb)
pixels.show()
def off_light_bar():
for i in range(5): set_led_color(i, (0, 0, 0))
pixels.show()
# ------------------------------------------------------------------------------
# Hardware Initialization
# ------------------------------------------------------------------------------
cs_pin = DigitalInOut(board.CE0)
dc_pin = DigitalInOut(board.D25)
reset_pin = DigitalInOut(board.D24)
spi = board.SPI()
disp = st7789.ST7789(spi, height=240, y_offset=80, rotation=180,
cs=cs_pin, dc=dc_pin, rst=reset_pin, baudrate=24000000)
button_A = DigitalInOut(board.D5); button_A.direction = Direction.INPUT
button_B = DigitalInOut(board.D6); button_B.direction = Direction.INPUT
button_U = DigitalInOut(board.D17); button_U.direction = Direction.INPUT
button_D = DigitalInOut(board.D22); button_D.direction = Direction.INPUT
button_L = DigitalInOut(board.D27); button_L.direction = Direction.INPUT
button_R = DigitalInOut(board.D23); button_R.direction = Direction.INPUT
button_C = DigitalInOut(board.D4); button_C.direction = Direction.INPUT
backlight = DigitalInOut(board.D26)
backlight.switch_to_output()
backlight.value = True
width, height = disp.width, disp.height
image = Image.new("RGB", (width, height))
draw = ImageDraw.Draw(image)
udlr_fill, udlr_outline = "#00FF00", "#00FFFF"
fnt = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 30)
fnt_2 = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 24)
pixels = neopixel.NeoPixel(board.D12, 5, pixel_order=neopixel.GRB)
sensor = AS7341(board.I2C())
sensor.led_current = 10
kit = MotorKit()
stepper_motor1 = kit.stepper1
stepper_motor2 = kit.stepper2
stepper_speed_home = 0.000002
full_steps_per_revolution = 200 * 8
microsteps_per_revolution = full_steps_per_revolution
possible_angles_slide = [45, 90, 135, 180, 225, 270]
GPIO.setup(20, GPIO.IN, pull_up_down=GPIO.PUD_UP)
GPIO.setup(18, GPIO.IN, pull_up_down=GPIO.PUD_UP)
# ------------------------------------------------------------------------------
# Startup Ritual & Main Menu Loop
# ------------------------------------------------------------------------------
clear_screen(disp, draw, image, width, height)
splash_screen()
centroids = load_centroids("skittle_centroids.json")
go_home(stepper_motor1)
go_home_slide(stepper_motor2)
go_to_zero(stepper_motor2, 0)
now_bucket = 0
menu_numer = 1
menu_title = "Main Menu"
menu_text = ["Sort All", "Scan X10", "Home Servos", "Utilities", "Exit"]
while True:
try:
draw_menu(disp, draw, image, width, height, menu_title, menu_text, selected_index=menu_numer)
if not button_U.value:
menu_numer -= 1
if menu_numer < 1: menu_numer = len(menu_text)
time.sleep(0.2)
elif not button_D.value:
menu_numer += 1
if menu_numer > len(menu_text): menu_numer = 1
time.sleep(0.2)
if not button_A.value or not button_C.value:
if menu_numer == 1:
empty_count = 0
while empty_count < 3:
now_bucket, last_scan = next_scan_and_release(now_bucket)
if now_bucket == 0: empty_count += 1
else: empty_count = 0
status_display(disp, draw, image, width, height, text=["Finished", "Sorting"])
time.sleep(0.5)
elif menu_numer == 2:
for _ in range(10):
now_bucket, last_scan = next_scan_and_release(now_bucket)
elif menu_numer == 3:
status_display(disp, draw, image, width, height, text=["Homing", "Servos"])
go_home(stepper_motor1)
go_home_slide(stepper_motor2)
go_to_zero(stepper_motor2, 0)
now_bucket = 0
elif menu_numer == 4:
sub_menu_title = "Utilities"
sub_menu_text = ["Scan+LogX5", "Rot. Wheel", "Slide +", "Slide -", "Back"]
sub_menu_numer = 1
while True:
draw_menu(disp, draw, image, width, height, sub_menu_title, sub_menu_text, selected_index=sub_menu_numer)
if not button_U.value:
sub_menu_numer -= 1
if sub_menu_numer < 1: sub_menu_numer = len(sub_menu_text)
time.sleep(0.2)
elif not button_D.value:
sub_menu_numer += 1
if sub_menu_numer > len(sub_menu_text): sub_menu_numer = 1
time.sleep(0.2)
if not button_A.value or not button_C.value:
if sub_menu_numer == 1:
log_file = "sensor_log.csv"
for _ in range(5):
now_bucket, last_scan = next_scan_and_release(now_bucket)
with open(log_file, mode='a', newline='') as f:
csv.writer(f).writerow([datetime.now().strftime("%Y-%m-%d %H:%M:%S")] + last_scan)
time.sleep(0.5)
elif sub_menu_numer == 2:
rotate_to_angle(stepper_motor1, 90)
stepper_motor1.release()
elif sub_menu_numer == 3:
now_bucket = go_to_bucket(stepper_motor2, now_bucket, now_bucket + 1)
elif sub_menu_numer == 4:
now_bucket = go_to_bucket(stepper_motor2, now_bucket, now_bucket - 1)
elif sub_menu_numer == 5:
break
elif menu_numer == 5:
status_display(disp, draw, image, width, height, text=["Exiting", "Program"])
break
except KeyboardInterrupt:
break
# Graceful Shutdown
stepper_motor1.release()
stepper_motor2.release()
turn_off_all()
print("Clean shutdown complete.")
#!/usr/bin/env python3
import json, math, csv, argparse
from collections import defaultdict, Counter
from pathlib import Path
CHANNELS = ["Violet", "Indigo", "Blue", "Cyan", "Green", "Yellow", "Orange", "Red"]
def _canonical_label(raw: str) -> str:
if not raw: return "unknown"
first = raw.strip().split()[0].lower()
alias = {"purple": "purple", "green": "green", "yellow": "yellow", "orange": "orange"}
return alias.get(first, first)
def _normalize(vec, mode="sum"):
if mode == "sum":
s = sum(vec)
return [v / s if s > 0 else 0.0 for v in vec]
elif mode == "l2":
s2 = math.sqrt(sum(v*v for v in vec))
return [v / s2 if s2 > 0 else 0.0 for v in vec]
else:
raise ValueError("Unknown normalize mode")
def _cosine_sim(a, b):
dot = sum(x*y for x, y in zip(a, b))
na = math.sqrt(sum(x*x for x in a))
nb = math.sqrt(sum(y*y for y in b))
return dot / (na*nb) if na > 0 and nb > 0 else 0.0
def load_tsv(path: Path):
rows = []
with path.open(newline="") as f:
reader = csv.DictReader(f, dialect=csv.excel_tab if "\t" in f.read(2048) else csv.excel)
f.seek(0)
reader = csv.DictReader(f)
reader.fieldnames = [h.strip() for h in reader.fieldnames]
for r in reader:
label = _canonical_label(r.get(reader.fieldnames[0], ""))
try:
vec = [float(r[c]) for c in CHANNELS]
rows.append((label, vec))
except Exception:
continue
return rows
def compute_centroids(labeled_rows, norm="sum"):
buckets = defaultdict(list)
for lbl, vec in labeled_rows:
if lbl != "unknown":
buckets[lbl].append(_normalize(vec, norm))
centroids = {}
for lbl, vecs in buckets.items():
if vecs:
mean = [sum(v[i] for v in vecs)/len(vecs) for i in range(len(CHANNELS))]
centroids[lbl] = _normalize(mean, "l2")
return centroids
def classify(vec, centroids, norm="sum"):
x = _normalize(_normalize(vec, norm), "l2")
sims = [(lbl, _cosine_sim(x, c)) for lbl, c in centroids.items()]
sims.sort(key=lambda t: t[1], reverse=True)
return sims[0][0], sims[0][1], sims
def save_centroids(centroids, path: Path):
path.write_text(json.dumps({lbl: list(vec) for lbl, vec in centroids.items()}, indent=2))
def load_centroids(path: Path):
data = json.loads(Path(path).read_text())
return {lbl: [float(x) for x in vec] for lbl, vec in data.items()}
#!/usr/bin/env python3
# ==============================================================================
# Emergency Recovery Script: motor_release.py
# Instantly un-energizes all motor coils if a Python script crashes.
# ==============================================================================
from adafruit_motorkit import MotorKit
import time
print("Connecting to Adafruit Motor HAT (0x60)...")
kit = MotorKit(address=0x60)
print("Releasing Motor 1 (Inspector Wheel)...")
kit.stepper1.release()
print("Releasing Motor 2 (Sorting Slide)...")
kit.stepper2.release()
time.sleep(0.1)
print("✓ All stepper motor coils released and de-energized.")
{
"yellow": [
0.0381, 0.0512, 0.0824, 0.1412,
0.3210, 0.5841, 0.6120, 0.3840
],
"green": [
0.0410, 0.0620, 0.0950, 0.2810,
0.6420, 0.5110, 0.4210, 0.2310
],
"purple": [
0.2810, 0.3410, 0.2910, 0.1210,
0.0850, 0.1120, 0.4120, 0.7120
],
"orange": [
0.0290, 0.0380, 0.0510, 0.0820,
0.1810, 0.4810, 0.6920, 0.4910
]
}