code Code Page

C0P1

the Protocol Droid

To say that artificial intelligence (AI) was not important in this project would be a gross understatement. Given the complexity of inter-hardware communication, the novelty of concepts such as Flask, and the sheer number of robots and props involved, AI was and had to be used. We even renamed Microsoft Copilot™, C0P1 (a pun on Star Wars' C3PO) our very own protocol droid. It helped us in so many aspects of the project. We are eternally grateful :-)
All existing code has been ported to a class-based template and any new code, developed using that template. For the LEGO Mindstorms EV3 robots, there is an EV3DEV and Pybricks MicroPython version. For the LEGO SPIKE Prime robot and props, there is a Pybricks version.
Versioning was implemented and all iterations can be found on OneDrive in the Versions folder.

Bluey, the Velociraptor

The below are the main code for the ball / signature following and sending of its JSON data to the Flask server endpoint:

# Follow signature
def follow_signature(self):

	pixy = self.hw.pixy
	steer = self.hw.move_steering

	# Read Pixy blocks
	try:
		status, blocks_raw = pixy.get_blocks(1, 1)
	except OSError:
		print("[PIXY] I2C error. Retrying ...")
		time.sleep(0.05)
		return
	
	# Send JSON to Flask server
	self.send_pixy_json(blocks_raw)
	# Filter blocks
	blocks_for_following = []

	if blocks_raw:
		for b in blocks_raw:
			if b.sig == TARGET_SIG:
				blocks_for_following.append(b)

	# Pick largest block
	block = self.get_largest_block(blocks_for_following)

	if block:
		# Compute steering error
		error = block.x_center - FRAME_CENTER
		turn = KP * error
		turn = max(min(turn, 100), -100)

		message = '[FOLLOW] x:' + str(block.x_center) + ', Turn:' + str(turn)
		print(message)
	
		# Drive forward while steering
		steer.on(-turn, SpeedPercent(20))

	else:
		print("[FOLLOW] Object not found")
		steer.off()
	
# Find largest block
def get_largest_block(self, blocks):

	if not blocks:
		return None
	return max(blocks, key=lambda b: b.width * b.height)
	
# Send Pixy2 JSON to Flask server endpoint
def send_pixy_json(self, blocks_raw):

	blocks_json = []

	if blocks_raw:
		for b in blocks_raw:
			print(vars(b))
			if b.sig == TARGET_SIG:
				blocks_json.append({
					"x": 316 - b.x_center,
					"y": 208 - b.y_center,            
					"w": b.width,
					"h": b.height,
					"sig": b.sig,
					"angle": getattr(b, "angle", None)
				})

	payload = {
		"device": self.name,
		"blocks": blocks_json,
		"vectors": [],
		"barcodes": []
	}

	try:
		requests.post(self.server.server + "/receive_frame", json=payload, timeout=0.2)
	except:
		pass		
			

The entire code can be downloaded here.

Brontie, the Brachiosaurus

The below are the main code for the leaves detection and eating, and dino lifting:

# Detect leaves
def detect_leaves(self):

	ir = self.hw.sensor_ir

	# heading: -25..25, distance: 0..100 (cm)
	heading, distance = ir.heading_and_distance(channel=1)

	message = 'Heading: ' + str(heading) + ' Distance: ' + str(distance)
	print(message)

	# Turn toward beacon (softened)
	self.turn(heading)

	# No distance reading, do nothing
	if distance is None or distance == 0:
		return False

	# Soft approach zone (40–30 cm)
	if 30 < distance <= 40:
		print("[LEAVES] Soft approach ...")
		self.hw.motor_move.on_for_seconds(SPEED_LOW, 0.3)
		return False

	# Eating zone (< 30 cm)
	if distance <= 30:
		print("[LEAVES] Leaves detected. Stopping ...")
		self.hw.motor_move.stop()
		self.lift_dino()
		return True

	return False

# Eat leaves
def eat_leaves(self):

	jaw = self.hw.motor_eat

	print("[EAT] Eating leaves ...")

	# Open jaw
	jaw.on_for_seconds(SPEED_LOW, DURATION_HIGH)
	# Pause
	time.sleep(3)
	# Close jaw
	jaw.on_for_seconds(-SPEED_LOW, DURATION_HIGH)
	print("[EAT] Done eating leaves ...")
	self.server.send_event("EVENT", "01")
	
	self.leaves_not_detected = False
	return self.leaves_not_detected

# Lift dinosaur
def lift_dino(self):

	left = self.hw.motor_lift_left
	right = self.hw.motor_lift_right
	move = self.hw.motor_move
	turn = self.hw.motor_turn

	print("[LIFT] Lifting Brontie ...")
	msg = "Standing on hind legs"
	self.server.send_event("STATUS", msg)

	# Lift up
	left.run_forever(speed_sp=-SPEED_LIFT)
	right.run_forever(speed_sp=-SPEED_LIFT)
	time.sleep(WAIT_LIFT)

	left.stop()
	right.stop()
	# Eat leaves while lifted
	self.eat_leaves()
	# Lower down
	left.run_forever(speed_sp=SPEED_LIFT)
	right.run_forever(speed_sp=SPEED_LIFT)
	time.sleep(WAIT_LIFT)

	left.stop()
	right.stop()
	move.stop()
	turn.stop()

	print("[LIFT] Done lifting Brontie ...")		
			

The entire code can be downloaded here for the new version and here for the original.

Hamster Ball, the gyrosphere

The below are the main code for the ball following and rotation adjustment:

# Move with motor offset
def move(self, direction, speed, rotation):
	# Convert motor angles to radians (relative to unit circle) i.e.: 0 degrees is East
	angle_A = radians(45)
	angle_B = radians(135)
	angle_C = radians(225)
	angle_D = radians(315)

	# Convert direction to radians
	direction_angle = radians(direction)
		
	# Calculate speed for motors based on position of IR ball
	speed_A = cos((direction_angle - angle_A)) * speed + rotation
	speed_B = cos((direction_angle - angle_B)) * speed + rotation
	speed_C = cos((direction_angle - angle_C)) * speed + rotation
	speed_D = cos((direction_angle - angle_D)) * speed + rotation

	# Run motors at the calculated speed
	self.hw.motor_a.run(speed_A)
	self.hw.motor_b.run(speed_B)
	self.hw.motor_c.run(speed_C)
	self.hw.motor_d.run(speed_D)

# Rotate smoothly
def smooth_rotation(self, desired_heading):
	# Initialise function constants
	# Proportional gain (smoothness control)
	Kp = 0.5 # Lower value is smoother, higher is snappier
	# Deadzone to prevent jitter near 0 degrees
	deadzone = 4 # degrees
	
	# Reset heading to 0 degrees i.e.: 'North'
	self.hub.imu.reset_heading(0)
	current_heading = self.hub.imu.heading()
	# Calculate how far desired & current are apart
	error = desired_heading - current_heading
	# Set error between -180 and 180 degrees
	if error > 180:
		error -= 360
	elif error < -180:
		error += 360    
	
	if abs(error) < deadzone:
		return 0

	# Proportional correction
	rotation_temp = Kp * error
	# Clamp to avoid extreme rotation speeds
	rotation = max(min(rotation_temp, 100), -100)

	print("Current:", current_heading,
		  "Desired:", desired_heading,
		  "Error:", error,
		  "Rotation:", rotation)

	return rotation
			

The entire code can be downloaded here for the new version and here for the original converted to Pybricks or here for the original from the SPIKE Prime app.

Indie, the Indominus Rex

The below is the main code for the action handler:

# Update action
def update_action(self):

	action_current = FORWARD_SLOW
	self.timer_action.reset()
	yield action_current

	while self.timer_action.time() < 800:
		yield

	action_current = STOP
	yield action_current

	while True:
		distance = self.hw.sensor_us.distance()
		action_new = STOP

		# Far: gentle random turn & growl
		if distance is not None and distance >= DIST_FAR:
			print("[DIST] Far")
			turn = urandom.choice([TURN_LEFT, TURN_RIGHT])
			action_current = Action(speed_drive=FORWARD_SLOW.speed_drive,
									steering=turn.steering)
			yield action_current
			self.timer_action.reset()
			while self.timer_action.time() < 800:
				yield
			self.pending_behaviour["sound"] = "GROWL"

		# Mid: forward fast & jaw movement
		elif distance is not None and DIST_MID <= distance < DIST_FAR:
			print("[DIST] Mid")
			action_current = FORWARD_FAST
			yield action_current
			self.pending_behaviour["motion_jaw"] = "open_close"
			self.timer_action.reset()
			while self.timer_action.time() < 800:
				yield
			self.pending_behaviour["sound"] = "ATTACK"

		# Near: slow forward & arms, then slow backward
		elif distance is not None and DIST_NEAR <= distance < DIST_MID:            
			print("[DIST] Near")
			action_current = FORWARD_SLOW
			yield action_current
			self.pending_behaviour["motion_arms"] = "up_down"

			action_current = BACKWARD_SLOW
			yield action_current
			self.timer_action.reset()
			while self.timer_action.time() < 800:
				yield
			self.pending_behaviour["sound"] = "WIN"

		else:
			# Very close or no reading: stop
			print("[DIST] Very close or None")
			action_current = STOP
			yield action_current
			self.timer_action.reset()
			while self.timer_action.time() < 200:
				yield

		self.timer_action.reset()
		while self.timer_action.time() < 100:
			yield
			

The entire code can be downloaded here for the new version and here for the original. The code is based on Laurens Valk's Gyro Boy given the complexity of self-balancing.

Jeep, the Jeep

The below are the main code for the RFID tag reading, line following and obstacle avoidance, as well as Flask server communication:

# Find RFID tags
def check_rfid(self):

	uid = self.hw.sensor_rfid.readUID()
	uid_str = None
	# Normal behaviour after startup
	if uid and uid != self.hw.last_uid:
		self.hw.last_uid = uid
		
		# TCP event
		uid_str = str(uid).strip()
		mapped = RFID_MAP.get(uid_str, "00")
		# Ignore stale UID
		if self.first_uid and mapped != "01":
			self.first_uid = False
			return
		
		self.server.send_event("RFID", mapped)
		print("[RFID] UID:", uid, "Mapped:", mapped)

	if uid is None:
		self.last_uid = None

# Follow the line
def follow_line(self):
	# MODIFY TO SUIT REQUIRED CODE
	# Declare function variables
	intensity_left = self.hw.sensor_colour_left.reflection()
	intensity_right = self.hw.sensor_colour_right.reflection()

	if intensity_left >= self.threshold and intensity_right >= self.threshold:
		# Both see white, move forward
		self.hw.drive_base.drive(SPEED_MEDIUM, 0)
	elif intensity_left < self.threshold and intensity_right >= self.threshold:
		# Left sensor sees black, turn left medium
		self.hw.drive_base.drive(SPEED_MEDIUM, -RATE_MEDIUM)
	elif intensity_left >= self.threshold and intensity_right < self.threshold:
		# Right sensor sees black, turn right medium
		self.hw.drive_base.drive(SPEED_MEDIUM, RATE_MEDIUM)
	elif intensity_left < self.threshold and intensity_right < self.threshold:
		# Both see 'black', move forward slowly
		self.hw.drive_base.drive(SPEED_LOW, 0)
	# Wait to avoid excessive loop execution
	wait(TIME_WAIT)

# Detect & avoid obstacle
def handle_obstacle(self):

	distance = self.hw.sensor_us.distance()
	now = self.clock.time() / 1000  # s

	if distance is None:
		return False  # Treat as no obstacle

	if distance < THRESHOLD_OBSTACLE:
		msg = "Obstacle:" + str(distance) + "mm"
		self.server.send_event("STATUS", msg)
		
		if self.obstacle_start_time is None:
			self.obstacle_start_time = now
			print("[OBS] Obstacle detected. Waiting ...")
		self.hw.stop()

		# If obstacle persists too long, try to drive around
		if not self.avoiding and (now - self.obstacle_start_time) > OBSTACLE_TIMEOUT:
			print("[OBS] Timeout. Attempting to drive around ...")
			self.avoiding = True
			# Simple avoidance: back up, turn, move forward a bit
			self.hw.drive_base.straight(-100)
			self.hw.drive_base.turn(urandom.choice([-60, 60]))
			self.hw.drive_base.straight(150)
			# Reset and let line-follow re-acquire
			self.obstacle_start_time = None
			self.avoiding = False
		return True
	else:
		self.obstacle_start_time = None
		return False
	
# Send event
def send_event(self, kind, detail):
	# Populate JSON payload
	payload = {
		"device": self.device,
		"type": kind,
		"value": detail
	}
	# Try sending JSON event to /event endpoint on Flask server
	try:
		r = requests.post(self.server + "/event", json=payload, headers={"Connection": "close"}
)
		r.close()
		print("[EVENT]", payload)

	except Exception as e:
		print("[HTTP] Sending event failed with error '", e, "'")
		wait(200)

# Receive command
def get_command(self):
	# Try receiving JSON command from /sent_command/ endpoint on Flask server
	try:
		r = requests.get(self.server + "/sent_command/" + self.device, headers={"Connection": "close"}
)
		data = r.json()
		r.close()

		return data.get("command sent")

	except Exception as e:
		print("[HTTP] Receiving command failed with error '", e, "'")
		wait(200)
		return None
			

The entire code can be downloaded here for the new version and here for the original.

Rexie, the T-Rex

The below are the main code for the temperature scanning and obstacle 'attacking':

# Scan left, centre, right
def heat_seek_step(self):

	left = self.hw.motor_left
	right = self.hw.motor_right
	tank = self.hw.move_tank
	jaw = self.hw.motor_jaw
	
	centre_temp = self.scan_direction(0)
	left_temp = self.scan_direction(-SCAN_ANGLE)
	right_temp = self.scan_direction(SCAN_ANGLE)

	# Find hottest direction
	temps = {
		"centre": centre_temp,
		"left": left_temp,
		"right": right_temp
	}

	direction = max(temps, key=temps.get)
	hottest = temps[direction]

	print("[SCAN] Temperatures found", temps, ". Hottest at", direction, hottest)
	msg = "Temp.:" + str(hottest) + "degC, Dir.:" + direction
	self.server.send_event("STATUS", msg)

	# Jaw animation
	# jaw.on_for_seconds(SpeedPercent(40), 0.2)
	# jaw.on_for_seconds(SpeedPercent(-40), 0.2)
	jaw.on_to_position(SpeedPercent(40), -90)
	jaw.on_to_position(SpeedPercent(40), 0)
	
	# Move based on direction
	if direction == "right":
		tank.on_for_seconds(SpeedPercent(-20), SpeedPercent(20), 0.3)
	elif direction == "left":
		tank.on_for_seconds(SpeedPercent(20), SpeedPercent(-20), 0.3)
	else:
		# Move forward toward heat
		tank.on_for_seconds(SpeedPercent(-30), SpeedPercent(-30), 1)
	
# Scan 
def scan_direction(self, angle):

	scanner = self.hw.motor_sensor        
	scanner.on_to_position(SpeedPercent(20), angle, brake=True, block=True)
	time.sleep(0.2)
	temperature = self.read_target()
	# Reset to centre
	scanner.on_to_position(SpeedPercent(20), 0, brake=True, block=True)
	return temperature

# Read ambient temperature
def read_ambient(self):

	sensor = self.hw.sensor_ir
	
	sensor.mode = 'AMBIENT-C'
	time.sleep(0.05)
	ambient = sensor.value(0) / 100.0
	print("[SCAN] Ambient temperature", ambient, "degC")
	return ambient

# Read target temperature
def read_target(self):

	sensor = self.hw.sensor_ir
	
	sensor.mode = 'TARGET-C'
	time.sleep(0.05)
	target = sensor.value(0) / 100.0
	print("[SCAN] Target temperature", target, "degC")
	return target

# Attack obstacle
def attack_obstacle(self):
	
	distance_cm = self.hw.sensor_us.distance_centimeters
	
	if distance_cm is None:
		return False  # Treat as no obstacle
	
	distance = distance_cm * 10

	jaw = self.hw.motor_jaw
	tank = self.hw.move_tank
	
	if distance < OBSTACLE_DISTANCE:
		print("[OBS] Obstacle detected at", distance, "mm")
		msg = "Prey:" + str(distance) + "mm"
		self.server.send_event("STATUS", msg)
		self.server.send_event("EVENT", "01")
		# Charge forward
		tank.on_for_seconds(SpeedPercent(-60), SpeedPercent(-60), 1)
		# Jaw snap
		# jaw.on_for_seconds(SpeedPercent(80), 0.2)
		# jaw.on_for_seconds(SpeedPercent(-80), 0.2)
		jaw.on_to_position(SpeedPercent(80), -90)
		jaw.on_to_position(SpeedPercent(80), 0)
		self.server.send_event("EVENT", "04")
		
		return True
		
	return False
			

The entire code can be downloaded here.

Thomas, the monorail

The below are the main code for the barcode reading, line following and sending of its JSON data to the Flask server endpoint:

# Find barcodes
def check_bcid(self, data):
	# Send JSON for dashboard
	if not data or not data.barcodes:
		self.hw.last_uid = None
		return

	# Take first barcode
	bc = data.barcodes[0]
	code_str = str(bc.code)
	
	if code_str == self.hw.last_uid:
		return  # Already processed

	# Map to orchestration
	self.hw.last_uid = code_str
	mapped = BC_MAP.get(code_str)
	
	if mapped:
		msg = self.name + ":BC:" + mapped
		print("[BARCODE] UID:", code_str, "Mapped:", mapped)
		self.server.send_event("BARCODE", mapped)
	else:
		print("[BARCODE] Unknown barcode", code_str)

# Follow line
def follow_line(self, data):
	# MODIFY TO SUIT REQUIRED CODE
	if not data or not data.vectors:
		print("[LINE] Line not found")
		self.hw.move_tank.stop()
		return
	
	# If vector found, calculate horizontal distance from middle
	if data.number_of_vectors > 0:
		dx = data.vectors[0].x1 - X_REF
		self.move(dx)

# Send Pixy2 JSON to Flask server endpoint
def send_pixy_json(self, data):

	vectors_json = []
	barcodes_json = []

	if data is not None:
		# If vectors found
		if data.vectors:
			for v in data.vectors:
				# Compute angle manually for Pixy2
				dx = v.x1 - v.x0
				dy = v.y1 - v.y0
				angle = math.degrees(math.atan2(dy, dx))
				# Compute vector length
				#length = math.sqrt(dx*dx + dy*dy)
				#dx /= length
				#dy /= length
				# Extend vector length
				#x0_ext = v.x0 * SCALE + OFFSET_X
				#y0_ext = v.y0 * SCALE + OFFSET_Y
				#x1_ext = v.x1 * SCALE + OFFSET_X
				#y1_ext = v.y1 * SCALE + OFFSET_Y

				vectors_json.append({
					#"x0": x0_ext,
					#"y0": y0_ext,
					#"x1": x1_ext,
					#"y1": y1_ext,
					"x0": v.x0,
					"y0": v.y0,
					"x1": v.x1,
					"y1": v.y1,
					"angle": round(angle, 2)
				})

		# If barcodes found
		if data.barcodes:
			for bc in data.barcodes:
				barcodes_json.append({
					"x": bc.x,
					"y": bc.y,
					"code": bc.code
				})

	payload = {
		"device": self.name,
		"blocks": [],
		"vectors": vectors_json,
		"barcodes": barcodes_json
	}

	try:
		requests.post(self.server.server + "/receive_frame", json=payload, timeout=0.2)
	except:
		pass   		
			

The entire code can be downloaded here.

Aaron and Pedro cave

The below are the main code for Aaron biting and Pedro flapping and lifting:

# Make Aaron jump out of cave, bite Indie & go back in cave
def bite(self):
	print("[AARON] Bite triggered!")

	m = self.hw.motor_aaron

	# Jump out
	m.run_angle(SPEED_MOTOR, ANGLE_OPEN)

	wait(WAIT_TIME * 5)

	# Bite motion
	m.run_angle(SPEED_MOTOR, -ANGLE_ROTATE)
	m.run_angle(SPEED_MOTOR, ANGLE_ROTATE)

	wait(WAIT_TIME * 5)

	# Return to cave
	m.run_angle(SPEED_MOTOR, -ANGLE_OPEN)

	# Broadcast event
	self.ble.broadcast("EVENT", "AAP_bite_done")

# Make Pedro flap wings, lift up & down
def flap_lift(self, mode):
	"""
	mode = "flap", "lift", or "flap_lift"
	"""

	speed = SPEED_MOTOR
	angle_flap = ANGLE_FLAP
	angle_lift = ANGLE_LIFT

	pedro_top = self.hw.motor_pedro_top
	pedro_bottom = self.hw.motor_pedro_bottom

	print("[PEDRO] Performing", mode)

	# Number of repetitions
	if mode == "flap":
		reps = 6
	elif mode == "lift":
		reps = 2
	elif mode == "flap_lift":
		reps = 2
	else:
		return

	# Initial movement
	if mode == "flap":
		pedro_top.run_target(speed, angle_flap)
	elif mode == "lift":
		pedro_bottom.run_target(speed, angle_lift)
	elif mode == "flap_lift":
		pedro_top.run_target(speed, angle_flap, wait=False)
		pedro_bottom.run_target(speed, angle_lift)

	# Repeated movement
	for _ in range(reps):
		if mode == "flap":
			pedro_top.run_angle(speed, angle_flap)
		elif mode == "lift":
			pedro_bottom.run_angle(speed, angle_lift)
		elif mode == "flap_lift":
			pedro_top.run_angle(speed, angle_flap, wait=False)
			pedro_bottom.run_angle(speed, angle_lift)

		# Toggle direction
		angle_flap = -angle_flap
		angle_lift = -angle_lift

	# Reset to neutral
	if mode == "flap":
		pedro_top.run_target(speed, 0)
	elif mode == "lift":
		pedro_bottom.run_target(speed, 0)
	elif mode == "flap_lift":
		pedro_top.run_target(speed, 0, wait=False)
		pedro_bottom.run_target(speed, 0)
			

The entire code can be downloaded here for the new version and here for the original.

Jurassic Kingdom gate

The below are the main code for the lights, gate doors and motor reset (which is common to all SPIKE Prime props):

# Start lights at frequency of change
def light_up(self, frequency):
	print('Turning lights on at', frequency, 'frequency ...')
	# If frequency low, choose high wait time in ms
	if frequency == 'low':
		wait_lights = randint(200,500)
	# If frequency high, choose low wait time in ms
	else:
		wait_lights = randint(100,200)
		
	# Run gate lights
	self.hw.motor_lights.run_angle(SPEED_MOTOR, ANGLE_LIGHTS) # Gate lights
	wait(wait_lights)
	
	self.hw.motor_lights.hold() # Gate lights        
	wait(WAIT_TIME)

 # Shake trees at frequency of change
def shake_trees(self, frequency):
	print('Shaking trees at', frequency, 'frequency ...')
	# If frequency low, choose high wait time in ms
	if frequency == 'low':
		wait_trees = randint(200,500)
	# If frequency high, choose low wait time in ms
	else:
		wait_trees = randint(100,200)
		
	# Rotate tree motors
	self.hw.motor_tree_left.run_angle(SPEED_MOTOR, ANGLE_ROTATE, wait=False) # Left tree motor
	self.hw.motor_tree_left.run_angle(SPEED_MOTOR, ANGLE_ROTATE) # Right tree motor
	wait(wait_trees)

	ANGLE_ROTATE *= -1

# Open or close gate
def open_close_gate(self, action):        
	speed = SPEED_MOTOR / 5
	if action == 'open':
		print('Opening gate ...')
		self.ble.broadcast("EVENT", "JKG_open")
		self.hw.motor_gate_left.run_angle(speed, self.angle_gate, wait=False)
		self.hw.motor_gate_right.run_angle(speed, self.angle_gate)
		#self.angle_gate = self.angle_gate * -1
		self.angle_gate *= -1
	
	elif action == 'close':
		print('Closing gate ...')
		self.ble.broadcast("EVENT", "JKG_close")
		self.hw.motor_gate_left.run_angle(speed, self.angle_gate, wait=False)
		self.hw.motor_gate_right.run_angle(speed, self.angle_gate)
		#self.angle_gate = self.angle_gate * -1
		self.angle_gate *= -1
	
# Reset gate & light motors
def reset(self, motors):
	# Initialize counter
	counter_motor = 0
	# Loop through motors array
	for angle in motors:
		# If a value is found in array
		if angle != '':
			print('Resetting motor', self.hw.list_motors[counter_motor], 'at', angle, 'degrees ...')
			# Look up motor, reset position to angle
			self.hw.list_motors[counter_motor].run_target(SPEED_MOTOR, angle)
		# Increment counter
		counter_motor += 1		
			

The entire code can be downloaded here for the new version and here for the original.

Moses and feeding set

The below are the main code for Moses lunging and the shark rotating:

# Make Moses lunge			
def moses_lunge(self):
	print("[MOSES] Lunge!")
	# Small turn to release elastic
	self.hw.motor_moses_lunge.run_angle(SPEED_MOTOR, ANGLE_LUNGE_RELEASE)
	self.lunge_done = True
	self.ble.broadcast("EVENT", "lunged")

# Make Moses wind back
def moses_wind_back(self):
	print("[MOSES] Winding back ...")
	#self.hw.motor_moses_wind.run_angle(SPEED_MOTOR, ANGLE_WIND_BACK)
	#self.hw.motor_moses_wind.run_angle(SPEED_MOTOR, -ANGLE_WIND_BACK)
	self.hw.motor_moses_lunge.run_angle(SPEED_MOTOR, -ANGLE_LUNGE_RELEASE)
	self.ble.broadcast("EVENT", "reset")

# Start shark rotating
def shark_start(self):
	print("[SHARK] Start swinging")
	self.shark_running = True
	self.ble.broadcast("EVENT", "start")

# Stop shark rotating
def shark_stop(self):
	print("[SHARK] Stop swinging")
	self.shark_running = False
	self.hw.motor_shark_turn.stop()
	self.ble.broadcast("EVENT", "stop")      
			

The entire code can be downloaded here.

Toilet from the iconic movie scene

The below are the main code for the toilet 'explosion' and the periodic method (which is similar for all SPIKE Prime props):

# Make toilet 'explode'			
def explode(self):
	self.ble.broadcast("CMD", "JKG_shhi")
	print("[TOILET] EXPLOSION triggered!")
	self.hw.motor_front.run_angle(SPEED_MOTOR, ANGLE_EXPLODE_FRONT)
	self.hw.motor_back.run_angle(SPEED_MOTOR, ANGLE_EXPLODE_BACK)
	self.exploded = True
	self.ble.broadcast("CMD", "TOI_stop")
	wait(3000)
	self.ble.broadcast("CMD", "JKG_wait")  

# Define functions run periodically
def periodic(self): # <--- Add required code        
	# If already exploded, do nothing
	if self.exploded:
		return

	dist = self.hw.sensor_us.distance()
	if dist is not None:
		print("[US] Distance:", dist)
		self.ble.broadcast("CMD", "TOI_open")
		self.ble.broadcast("CMD", "MOF_start")
		self.ble.broadcast("CMD", "JKG_shlo")

		# First detection
		if not self.waiting_for_confirm and dist < DIST_TRIGGER:
			print("[TOILET] First detection - Waiting for confirm ...")
			self.waiting_for_confirm = True
			self.confirm_start_time = self.clock.time()

		# Confirm detection
		if self.waiting_for_confirm:
			now = self.clock.time()

			if dist < DIST_CONFIRM:
				print("[TOILET] Confirmed — EXPLODING!")
				self.explode()
				self.waiting_for_confirm = False
				#self.ble.broadcast("EVENT", "TOI_open")
				#self.ble.broadcast("CMD", "TOI_open")

			elif now - self.confirm_start_time > 1000:
				print("[TOILET] Confirm timeout — Cancelling ...")
				self.waiting_for_confirm = False

	# Backup-confirm US detection
	dist_back = self.hw.sensor_us_back.distance()
	if dist_back is not None:
		print("[US BACK] Distance:", dist_back)
		self.ble.broadcast("CMD", "TOI_open")
		self.ble.broadcast("CMD", "MOF_start")
		self.ble.broadcast("CMD", "JKG_shlo")

		# First detection
		if not self.waiting_for_confirm and dist_back < DIST_TRIGGER:
			print("[TOILET] First back detection - Waiting for confirm ...")
			self.waiting_for_confirm = True
			self.confirm_start_time = self.clock.time()

		# Confirm detection
		if self.waiting_for_confirm:
			now_back = self.clock.time()

			if dist_back < DIST_CONFIRM:
				print("[TOILET] Confirmed — EXPLODING!")
				self.explode()
				self.waiting_for_confirm = False
				#self.ble.broadcast("EVENT", "TOI_open")
				#self.ble.broadcast("STATUS", "exploding")

			elif now_back - self.confirm_start_time > 1000:
				print("[TOILET] Confirm timeout — Cancelling ...")
				self.waiting_for_confirm = False
			

The entire code can be downloaded here.

Volcano

The below are the main code for the BLE broadcast and observe methods (which are common to all SPIKE Prime props) and the rumbling and erupting:

# Send event
def broadcast(self, kind, value):
	# Populate message
	msg = self.device_name + ":" + kind + ":" + value

	payload = msg[:21] # BLE advertisement limit of 21 char

	print("[BLE] Sending payload '", payload, "'")
	# Send 3 times for reliability
	for i in range(3):
		self.hub.ble.broadcast(payload)
		wait(POLL_INTERVAL)

# Receiving event    
def observe(self):
	for ch in self.observe_channels:
		msg = self.hub.ble.observe(ch)
		
		if msg:
			print("[BLE] Receiving message '", msg, "'")
			return msg
	
	return None

# Make volcano 'erupt'
def erupt(self):
	print("[VOLCANO] ERUPTION triggered!")
	self.erupted = True
	self.rumbling = False
	self.ble.broadcast("EVENT", "erupted")
	self.ble.broadcast("CMD", "JKG_shhi")

# Make the IR remote motor move back & forth
def remote_pulse(self, speed):
self.hw.motor_remote.run_angle(speed, ANGLE_REMOTE_PULSE)
self.hw.motor_remote.run_angle(speed, -ANGLE_REMOTE_PULSE)

# Make volcano rumble
def rumble(self):
	print("[VOLCANO] Rumble mode")
	self.rumbling = True
	self.erupted = False		
			

The entire code can be downloaded here.