From 5b64fbca5ac439c473f6222aba8d2d55f9898a66 Mon Sep 17 00:00:00 2001 From: videopixil Date: Thu, 7 May 2026 18:22:59 -0400 Subject: [PATCH 1/3] BLE beacon ears QTPy code to make into a BLE receiver --- BLE_Beacon_Ears/code.py | 792 ++++++++++++++++++++++++ BLE_Beacon_Ears/magicband_protocol.py | 373 ++++++++++++ BLE_Beacon_Ears/pixel_zones.py | 161 +++++ BLE_Beacon_Ears/renderer.py | 839 ++++++++++++++++++++++++++ 4 files changed, 2165 insertions(+) create mode 100755 BLE_Beacon_Ears/code.py create mode 100644 BLE_Beacon_Ears/magicband_protocol.py create mode 100644 BLE_Beacon_Ears/pixel_zones.py create mode 100644 BLE_Beacon_Ears/renderer.py diff --git a/BLE_Beacon_Ears/code.py b/BLE_Beacon_Ears/code.py new file mode 100755 index 000000000..b3673f2d0 --- /dev/null +++ b/BLE_Beacon_Ears/code.py @@ -0,0 +1,792 @@ +# SPDX-FileCopyrightText: 2026 Pedro Ruiz for Adafruit Industries +# +# SPDX-License-Identifier: MIT +'''MagicBand+ Beacon Ears main loop. + +Scans for Disney BLE adverts and renders matching commands on stereo +NeoPixel Jewels. Two logical zones (left ear, right ear) are rendered +with stereo phase offsets to give static colors a gentle out-of-phase +breathing animation and rotations a left-leads-right sweep. + +To switch from real Jewels to onboard NeoPixel for testing without the +ear hardware, flip _USE_ONBOARD to True below. +''' +# Target: Adafruit QT Py ESP32-S3 - the BLE Beacon Ears +import math +import random +import time + +import _bleio +import board +import digitalio +import neopixel +from adafruit_debouncer import Button + +import battery as battery_mod +import magicband_protocol +import pixel_zones +import renderer + +# Flip this to True to test on the QT Py's onboard pixel without jewels. +_USE_ONBOARD = False + +# 15 fps target -> ~66ms per frame. +_TARGET_FPS = 15 +_FRAME_BUDGET_S = 1.0 / _TARGET_FPS + +# BLE scan timing. +# interval and window equal means continuous scanning. Each scan burst is +# 40ms long which is our per-frame scan budget, then we render and sleep +# for the rest of the frame. This preserves smooth 15fps animation during +# active commands. +_SCAN_INTERVAL_S = 0.04 +_SCAN_WINDOW_S = 0.04 + +_MIN_RSSI = -90 +_DEDUP_WINDOW_S = 2.5 + +# Custom-protocol triggers (Ears Brightness) are broadcast by the CLUE +# for ~3 seconds per fire. The normal dedup window is 2.5s, so the same +# trigger packet would be accepted a second time near the end of the +# CLUE's broadcast. Use a longer cooldown specifically for these +# packets to guarantee one-fire-per-press behavior. +_TRIGGER_COOLDOWN_S = 4.0 + +# Disney park location beacons broadcast continuously - not just at the +# Fab 50 statues but also throughout attractions. They blast many packets +# per second per beacon. This long cooldown ensures the swirl animation +# fires only occasionally as you walk through the park, not every few +# seconds. +_STATUE_COOLDOWN_S = 30.0 + +# NeoPixel brightness presets cycled by a long-press of the BOOT button. +# Index 0 is the default used at startup. Keep presets modest - NeoPixels +# look much brighter on video cameras than to the eye, and at night even +# 0.02 reads clearly on the headband. +_BRIGHTNESS_PRESETS = (0.02, 0.04, 0.08) +# How long the BOOT button must be held to count as a long press. +# Short press cycles brightness preset (the more frequently-used action). +# Long press shows the battery level (less frequent, requires intent). +_LONG_PRESS_S = 0.5 +_BRIGHTNESS_FLASH_DURATION_S = 0.6 + +# USB-presence detection threshold (raw QT Py voltage reading). +# When charging via USB, the BFF charger pulls cell voltage up to ~4.20V +# which the QT Py reads as ~4.85V raw (uncalibrated ADC). True battery +# voltage even at 100% charge sits below 4.20V (typically 4.10-4.15V +# rest voltage), so a raw reading > 4.75V is a reliable indicator of +# USB power being present. +# +# We use this to gate the battery display: WS2812 NeoPixel timing gets +# corrupted on battery power (voltage sag during current spikes makes +# green data display as red), so we only show the full battery animation +# when on USB. On battery, BOOT short-press / Ears Battery shows a brief +# yellow flash instead - intuitive "plug in to check" feedback. +_USB_PRESENT_V_RAW = 4.75 +_UNAVAILABLE_FLASH_DURATION_S = 0.6 +_UNAVAILABLE_FLASH_COLOR = (255, 180, 0) # yellow - safe color on battery + +# On-demand battery display (BOOT short press or CLUE Ears Battery). +# 3-second animated display showing battery level on both jewels. +_BATTERY_DISPLAY_FILL_S = 1.2 +_BATTERY_DISPLAY_HOLD_S = 1.5 +_BATTERY_DISPLAY_FADE_S = 0.3 +_BATTERY_DISPLAY_DURATION_S = (_BATTERY_DISPLAY_FILL_S + + _BATTERY_DISPLAY_HOLD_S + + _BATTERY_DISPLAY_FADE_S) + +# Voltage thresholds tuned to QT Py's RAW (uncalibrated) ADC readings. +# The ESP32-S3 ADC reads ~0.65V higher than actual, so these numbers +# look high relative to true LiPo voltages. They are calibrated to +# match what the QT Py reports when battery is at the corresponding +# real charge level. +# +# QT Py raw reading | Real cell voltage | Visual +# ----------------- | ----------------- | ------ +# > 4.85V | > 4.20V (full) | 6 green +# > 4.70V | > 4.05V (~80%) | 5 green +# > 4.55V | > 3.92V (~50%) | 4 yellow +# > 4.40V | > 3.79V (~30%) | 3 yellow +# > 4.25V | > 3.65V (~15%) | 2 orange +# > 4.10V | > 3.55V (~5%) | 1 orange +# below | < 3.55V critical | red pulse +# +# The right-hand "real voltage" column was derived from Pedro's +# reference reading (Feather S2 voltage monitor showed 4.17V while +# QT Py raw read 4.83V). This is approximate; real-world calibration +# may shift these thresholds by ~0.05-0.10V. +_BATTERY_LEVELS = ( + (4.85, 6, (0, 180, 0)), # full green + (4.70, 5, (0, 180, 0)), # green + (4.55, 4, (180, 180, 0)), # yellow + (4.40, 3, (180, 180, 0)), # yellow + (4.25, 2, (220, 110, 0)), # orange + (4.10, 1, (220, 110, 0)), # orange +) +_BATTERY_CRITICAL_RGB = (220, 0, 0) +_BATTERY_UNKNOWN_RGB = (120, 0, 120) +_OUTER_RING_COUNT = 6 + +# Solo mode: lets the ears cycle through a curated showpiece reel without +# a CLUE remote nearby. Designed for solo demos and pickup video shots. +# Triple-press the BOOT button to toggle, double-press to skip the current +# showpiece, single-press still cycles brightness, long-press still shows +# battery. Real BLE packets interrupt solo cleanly so park interaction +# still works while solo is enabled - wearer doesn't manage state. +_SOLO_BREATH_S = 0.5 # idle gap between showpieces +_SOLO_DEFAULT_DURATION_S = 9.5 # fallback when a showpiece has no timing + +# Solo enter/exit indicator pulses. Sized to read clearly on camera so +# the moment of toggling is obvious in B-roll. +_SOLO_INDICATOR_ENTER_S = 0.6 # white burst: 200ms hold + 400ms fade +_SOLO_INDICATOR_EXIT_S = 0.7 # cool blue triangle pulse +_SOLO_EXIT_RGB = (40, 100, 220) + +# Debouncer multi-press timing windows. +# short_duration_ms is how long after a release the debouncer waits before +# committing a multi-press count. 350ms is a common sweet spot - tight +# enough that single-press brightness cycling still feels responsive, +# wide enough that comfortable triple-presses register reliably. +_BUTTON_SHORT_MS = 350 +_BUTTON_LONG_MS = int(_LONG_PRESS_S * 1000) + + +def _build_showpieces(): + '''Construct the solo-mode showpiece reel as parsed command dicts. + + Each entry goes through magicband_protocol.parse() so the renderer + treats it identically to a live BLE packet. Park captures from + Epcot/MK can be added by appending raw payloads here. + ''' + mp = magicband_protocol + # 9.5s timing byte: scaler off, no fade, time_val=2 -> 1.5*2 + 6.5 + timing_short = 0x02 + showpieces = ( + # Headliners: firmware-baked rainbow rotation + ("Taste the Rainbow", mp.parse( + bytes.fromhex("e100e90c000f0f5d465bf00532374830b0"))), + # 5-palette cycle, cruise-line warm rainbow + ("DCL Rainbow", mp.parse(bytes(( + 0xE1, 0x00, 0xE9, 0x0C, 0x00, 0x05, 0x0F, + 0x15, # Red + 0x13, # Orange + 0x10, # Off Yellow + 0x19, # Green + 0x07, # Deep Purple + 0xB0)))), + # Dual-color combos + ("Red+Cyan", mp.parse( + mp.build_dual_color(0x15, 0x00, timing=timing_short))), + ("Purple+Yellow", mp.parse( + mp.build_dual_color(0x07, 0x10, timing=timing_short))), + # Five-color combos + ("5C Rainbow", mp.parse( + mp.build_five_color( + center=0x10, top_left=0x15, bottom_left=0x13, + bottom_right=0x19, top_right=0x07, + timing=timing_short))), + ("5C Sunset", mp.parse( + mp.build_five_color( + center=0x13, top_left=0x08, bottom_left=0x14, + bottom_right=0x0F, top_right=0x10, + timing=timing_short))), + # 6-bit RGB demo - deep "Disney blue" the palette can't hit + ("RGB6 Disney Blue", mp.parse( + mp.build_six_bit_color(8, 24, 60, timing=timing_short))), + # --- Disney park show captures (Epcot, April 2026) --- + # Color labels match what was visually observed during capture. + # Payloads come straight from BLE listen logs. Long-format + # variants (E9 10/13, EA 14) aren't fully decoded yet so the + # renderer falls back to a generic park-show pulse with a + # payload-derived primary hue. Capture 6 (purple orange) had + # no obvious show packet in its log so it's omitted here. + ("Show: blue", mp.parse(bytes.fromhex( + "e91300b60f404458f44882d06519d146060a307bff"))), + ("Show: white light blue", mp.parse(bytes.fromhex( + "e90800b50fa8a3b3b3a3"))), + ("Show: rainbow", mp.parse(bytes.fromhex( + "e90400010efd"))), + ("Show: blue green", mp.parse(bytes.fromhex( + "e90800f40fa0a4b9b9a4"))), + ("Show: white sparkling", mp.parse(bytes.fromhex( + "ea140100410f434858f44882d06520d1460208307b40"))), + ("Show: orange red sparkle", mp.parse(bytes.fromhex( + "e910000f0f545d58f44882d146090ad06528"))), + ) + return showpieces + + +_SHOWPIECES = _build_showpieces() + + +def _pick_showpiece_idx(last_idx): + '''Pick a random showpiece index that isn't the same as last_idx.''' + if len(_SHOWPIECES) <= 1: + return 0 + candidates = [i for i in range(len(_SHOWPIECES)) if i != last_idx] + return random.choice(candidates) + + +def _render_solo_enter(zones, t): + '''White burst indicator: full hold for 200ms, fade over 400ms.''' + if t < 0.2: + envelope = 1.0 + else: + envelope = max(0.0, 1.0 - (t - 0.2) / 0.4) + color = (int(255 * envelope),) * 3 + for zone in zones: + zone.fill(color) + + +def _render_solo_exit(zones, t): + '''Cool blue pulse indicator: triangle envelope over 700ms.''' + frac = t / _SOLO_INDICATOR_EXIT_S + if frac < 0.4: + envelope = frac / 0.4 + else: + envelope = max(0.0, (1.0 - frac) / 0.6) + color = (int(_SOLO_EXIT_RGB[0] * envelope), + int(_SOLO_EXIT_RGB[1] * envelope), + int(_SOLO_EXIT_RGB[2] * envelope)) + for zone in zones: + zone.fill(color) + + +def _battery_level_match(voltage): + '''Return (count, rgb) for the given voltage, or None for critical.''' + if voltage is None: + return None + for threshold, count, rgb in _BATTERY_LEVELS: + if voltage >= threshold: + return (count, rgb) + return None # below lowest threshold = critical + + +def _render_battery_critical(zones, voltage, t): + '''Single-pixel critical/unknown battery animation. + + Used when voltage is too low to map to a level, or when no reading + is available. A pixel swirls around the outer ring for 1s, then + settles on a fixed location. + ''' + if voltage is None: + color = _BATTERY_UNKNOWN_RGB + else: + # Critical - pulsing red with a faster animation + pulse = 0.4 + 0.6 * math.sin(2 * math.pi * t * 2.5) + color = (int(_BATTERY_CRITICAL_RGB[0] * pulse), + int(_BATTERY_CRITICAL_RGB[1] * pulse), + int(_BATTERY_CRITICAL_RGB[2] * pulse)) + for zone in zones: + zone.set_led(0, (0, 0, 0)) + if t < 1.0: + head = int(t / 0.17) % _OUTER_RING_COUNT + for i in range(1, zone.count): + zone.set_led(i, color if (i - 1) == head else (0, 0, 0)) + else: + for i in range(1, zone.count): + zone.set_led(i, color if i == 1 else (0, 0, 0)) + + +def _battery_phase1_swirl(zones, t, fill_end, target_count, rgb): + '''Phase 1: swirling fill - leading-edge sweeps the ring.''' + sweep_pos = (t / fill_end) * _OUTER_RING_COUNT + for zone in zones: + zone.set_led(0, (0, 0, 0)) + for i in range(1, zone.count): + ring_idx = i - 1 + distance_since_passed = sweep_pos - (ring_idx + 1) + if distance_since_passed < 0: + zone.set_led(i, (0, 0, 0)) + elif distance_since_passed < 0.5: + if ring_idx + 1 <= target_count: + zone.set_led(i, rgb) + else: + fade = max(0.0, 1.0 - distance_since_passed * 2) + zone.set_led(i, (int(60 * fade),) * 3) + else: + zone.set_led(i, rgb if ring_idx + 1 <= target_count else (0, 0, 0)) + + +def _battery_phase2_hold(zones, hold_t, target_count, rgb): + '''Phase 2: gentle pulse on filled-in pixels.''' + pulse = 0.9 + 0.1 * math.sin(2 * math.pi * hold_t / 1.2) + pulsed = (int(rgb[0] * pulse), int(rgb[1] * pulse), int(rgb[2] * pulse)) + for zone in zones: + zone.set_led(0, (0, 0, 0)) + for i in range(1, zone.count): + zone.set_led(i, pulsed if (i - 1) + 1 <= target_count else (0, 0, 0)) + + +def _battery_phase3_fade(zones, fade_t, target_count, rgb): + '''Phase 3: smoothly fade filled pixels to black.''' + fade_fraction = max(0.0, 1.0 - fade_t / _BATTERY_DISPLAY_FADE_S) + if fade_fraction < 0.1: + faded = (0, 0, 0) + else: + faded = (int(rgb[0] * fade_fraction), + int(rgb[1] * fade_fraction), + int(rgb[2] * fade_fraction)) + for zone in zones: + zone.set_led(0, (0, 0, 0)) + for i in range(1, zone.count): + zone.set_led(i, faded if (i - 1) + 1 <= target_count else (0, 0, 0)) + + +def _render_battery_display(zones, voltage, t): + '''Paint both jewels with an animated battery level indicator. + + - Phase 1 (fill): a single leading-edge pixel sweeps around the outer + ring, leaving lit pixels behind it up to the target count. + - Phase 2 (hold): all active pixels glow with a gentle pulse. + - Phase 3 (fade): all pixels smoothly dim to black. + ''' + matched = _battery_level_match(voltage) + + # Diagnostic: log path on first frame + if t < 0.05: + v_str = f"{voltage:.3f}V" if voltage is not None else "None" + if matched is None: + print(f"[battery render] {v_str} -> CRITICAL/unknown path") + else: + print(f"[battery render] {v_str} -> {matched[0]} pixels") + + if matched is None: + _render_battery_critical(zones, voltage, t) + return + + target_count, rgb = matched + fill_end = _BATTERY_DISPLAY_FILL_S + hold_end = fill_end + _BATTERY_DISPLAY_HOLD_S + if t < fill_end: + _battery_phase1_swirl(zones, t, fill_end, target_count, rgb) + elif t < hold_end: + _battery_phase2_hold(zones, t - fill_end, target_count, rgb) + else: + _battery_phase3_fade(zones, t - hold_end, target_count, rgb) + + +# Custom sub-protocol for remote-triggered actions. +# Uses the 0x0183 Disney CID but with a first byte (0xAA) that is +# neither MagicBand+ (E1/E2/CC) nor Starlight Wand (CF/C0). Real bands +# and wands ignore packets they don't recognize, so this is safe. +# Command byte: +# 0x01 = show battery level +# 0x03 = cycle brightness preset +# 0x04 = find me +# 0x05 = statue animation preview +_REMOTE_BATTERY_PACKET = bytes.fromhex("aa4201") +_REMOTE_BRIGHTNESS_PACKET = bytes.fromhex("aa4203") +_REMOTE_FIND_PACKET = bytes.fromhex("aa4204") +_REMOTE_STATUE_PACKET = bytes.fromhex("aa4205") + + +def _is_remote_battery_trigger(payload): + return bytes(payload[:3]) == _REMOTE_BATTERY_PACKET + + +def _is_remote_brightness_trigger(payload): + return bytes(payload[:3]) == _REMOTE_BRIGHTNESS_PACKET + + +def _is_remote_find_trigger(payload): + return bytes(payload[:3]) == _REMOTE_FIND_PACKET + + +def _is_remote_statue_trigger(payload): + return bytes(payload[:3]) == _REMOTE_STATUE_PACKET + + +def _extract_disney_payload(ad_bytes): + '''Walk a BLE advert and extract the 0x0183 manufacturer payload.''' + i = 0 + while i < len(ad_bytes): + length = ad_bytes[i] + if length == 0 or i + 1 + length > len(ad_bytes): + break + ad_type = ad_bytes[i + 1] + if ad_type == 0xFF and length >= 3: + cid = ad_bytes[i + 2] | (ad_bytes[i + 3] << 8) + if cid == magicband_protocol.DISNEY_CID: + return bytes(ad_bytes[i + 4:i + 1 + length]) + i += 1 + length + return None + + +def _log_command(label, rssi, raw): + now = time.monotonic() + print(f"[{now:8.2f}] rssi={rssi:>4} {label}") + print(f" raw={raw.hex()}") + + + +class _RuntimeState: + '''All mutable state for the receiver main loop in one object. + + Bundling state into a single object keeps main() under the lint limit + for local variables and makes the per-frame helpers cleanly testable. + Plain data holder, no methods, like a struct. + ''' + # pylint: disable=too-many-instance-attributes + def __init__(self): + # Active rendered animation + self.active_state = None + self.active_started_at = 0.0 + # Dedup tracking for incoming BLE packets + self.last_payload = None + self.last_payload_at = 0.0 + # Brightness flash UI feedback + self.brightness_flash_until = 0.0 + self.brightness_flash_level = 0 + self.brightness_idx = [0] # list-wrapped for mutation reasons + # Trigger cooldowns to suppress repeated remote commands + self.last_trigger_time = 0.0 + self.last_statue_trigger = 0.0 + # On-demand battery display state + self.battery_display_until = 0.0 + self.battery_display_started_at = 0.0 + self.battery_display_voltage = None + # Yellow "plug in" flash (battery display unavailable) + self.unavailable_flash_until = 0.0 + self.unavailable_flash_started_at = 0.0 + # Find Me beacon mode + self.find_mode_until = 0.0 + self.find_mode_started_at = 0.0 + # Periodic battery voltage logging + self.last_battery_log = 0.0 + # Solo mode (curated showpiece reel) + self.solo_mode = False + self.solo_state = None + self.solo_label = "" + self.solo_idx = -1 + self.solo_started_at = 0.0 + self.solo_indicator_until = 0.0 + self.solo_indicator_started_at = 0.0 + self.solo_indicator_kind = None # 'enter' or 'exit' + + +_BATTERY_LOG_INTERVAL_S = 60.0 # log raw voltage once per minute + + +def _handle_button(state, button, batt, pixels): + '''Process one tick of BOOT-button input. + + Single-press: brightness cycle. Double-press (in solo): skip + showpiece. Triple-press: toggle solo mode. Long-press: battery. + ''' + button.update() + if button.long_press: + batt.update(force=True) + state.battery_display_voltage = batt.voltage + v_str = (f"{state.battery_display_voltage:.3f}V" + if state.battery_display_voltage is not None else "None") + now = time.monotonic() + if (state.battery_display_voltage is not None + and state.battery_display_voltage > _USB_PRESENT_V_RAW): + print(f"[battery trigger BOOT] {v_str} (USB - showing display)") + state.battery_display_until = now + _BATTERY_DISPLAY_DURATION_S + state.battery_display_started_at = now + else: + print(f"[battery trigger BOOT] {v_str} (battery - yellow flash)") + state.unavailable_flash_until = now + _UNAVAILABLE_FLASH_DURATION_S + state.unavailable_flash_started_at = now + elif button.short_count >= 3: + # Triple-press toggles solo mode. + now = time.monotonic() + state.solo_indicator_started_at = now + if state.solo_mode: + state.solo_mode = False + state.solo_state = None + state.solo_indicator_kind = "exit" + state.solo_indicator_until = now + _SOLO_INDICATOR_EXIT_S + print("[solo] exiting") + else: + state.solo_mode = True + state.solo_idx = -1 + state.solo_state = None + state.solo_indicator_kind = "enter" + state.solo_indicator_until = now + _SOLO_INDICATOR_ENTER_S + print("[solo] entering") + elif button.short_count == 2 and state.solo_mode: + state.solo_state = None + print("[solo] skipping to next showpiece") + elif button.short_count == 1: + state.brightness_idx[0] = ( + (state.brightness_idx[0] + 1) % len(_BRIGHTNESS_PRESETS)) + new_b = _BRIGHTNESS_PRESETS[state.brightness_idx[0]] + pixels.set_brightness(new_b) + state.brightness_flash_until = ( + time.monotonic() + _BRIGHTNESS_FLASH_DURATION_S) + state.brightness_flash_level = state.brightness_idx[0] + + +def _check_battery_log(state, batt): + '''Throttled battery voltage logger - one line per minute.''' + batt.update() + now = time.monotonic() + if (batt.voltage is not None + and now - state.last_battery_log >= _BATTERY_LOG_INTERVAL_S): + state.last_battery_log = now + level = _battery_level_match(batt.voltage) + level_str = f"{level[0]} pixels" if level else "CRITICAL" + print(f"[battery] raw={batt.voltage:.3f}V -> {level_str}") + + +def _render_brightness_flash(zones, state, frame_t): + '''Brightness-cycle confirmation flash on the outer ring.''' + flash_t = (_BRIGHTNESS_FLASH_DURATION_S + - (state.brightness_flash_until - frame_t)) + frac = flash_t / _BRIGHTNESS_FLASH_DURATION_S + if frac < 0.15: + envelope = frac / 0.15 + elif frac < 0.65: + envelope = 1.0 + else: + envelope = max(0.0, (1.0 - frac) / 0.35) + count = state.brightness_flash_level + 1 + lit = (int(255 * envelope),) * 3 + for zone in zones: + zone.set_led(0, (0, 0, 0)) + for i in range(1, zone.count): + zone.set_led(i, lit if (i - 1) < count else (0, 0, 0)) + + +def _render_unavailable_flash(zones, state, frame_t): + '''Yellow center pulse - "plug in to check battery".''' + flash_t = frame_t - state.unavailable_flash_started_at + frac = flash_t / _UNAVAILABLE_FLASH_DURATION_S + envelope = (frac * 2) if frac < 0.5 else max(0.0, 2 * (1 - frac)) + color = (int(_UNAVAILABLE_FLASH_COLOR[0] * envelope), + int(_UNAVAILABLE_FLASH_COLOR[1] * envelope), + int(_UNAVAILABLE_FLASH_COLOR[2] * envelope)) + for zone in zones: + zone.set_led(0, color) + for i in range(1, zone.count): + zone.set_led(i, (0, 0, 0)) + + +def _render_active_state(zones, state, frame_t): + '''Render the currently active animation, with crash safety.''' + t = frame_t - state.active_started_at + try: + state.active_state["render"](zones, t) + except Exception as err: # pylint: disable=broad-except + print(f"RENDER ERROR at t={t:.2f}s: " + f"{type(err).__name__}: {err}") + state.active_state = None + renderer.render_idle(zones) + + +def _render_solo_cycle(zones, state, frame_t): + '''Render one frame of solo mode\'s curated showpiece reel.''' + if state.solo_state is None: + state.solo_idx = _pick_showpiece_idx(state.solo_idx) + state.solo_label, parsed = _SHOWPIECES[state.solo_idx] + state.solo_state = renderer.for_command(parsed) + state.solo_started_at = frame_t + print(f"[solo] now playing: {state.solo_label}") + duration = state.solo_state["duration_s"] or _SOLO_DEFAULT_DURATION_S + t = frame_t - state.solo_started_at + if t >= duration + _SOLO_BREATH_S: + state.solo_state = None + renderer.render_idle(zones) + elif t >= duration: + renderer.render_idle(zones) + else: + try: + state.solo_state["render"](zones, t) + except Exception as err: # pylint: disable=broad-except + print(f"SOLO RENDER ERROR ({state.solo_label}) at " + f"t={t:.2f}s: {type(err).__name__}: {err}") + state.solo_state = None + renderer.render_idle(zones) + + +def _render_frame(zones, state, pixels): + '''Render one frame, picking the highest-priority active source.''' + frame_t = time.monotonic() + if 0.0 < state.find_mode_until <= frame_t: + pixels.set_brightness(_BRIGHTNESS_PRESETS[state.brightness_idx[0]]) + state.find_mode_until = 0.0 + print("[find me] animation complete, brightness restored") + + if frame_t < state.find_mode_until: + t = frame_t - state.find_mode_started_at + find_state = renderer.for_command({"kind": "find_me"}) + if find_state is not None: + find_state["render"](zones, t) + elif frame_t < state.solo_indicator_until: + t = frame_t - state.solo_indicator_started_at + if state.solo_indicator_kind == "enter": + _render_solo_enter(zones, t) + else: + _render_solo_exit(zones, t) + elif frame_t < state.brightness_flash_until: + _render_brightness_flash(zones, state, frame_t) + elif frame_t < state.unavailable_flash_until: + _render_unavailable_flash(zones, state, frame_t) + elif frame_t < state.battery_display_until: + t = frame_t - state.battery_display_started_at + _render_battery_display(zones, state.battery_display_voltage, t) + elif state.active_state is not None: + _render_active_state(zones, state, frame_t) + elif state.solo_mode: + _render_solo_cycle(zones, state, frame_t) + else: + renderer.render_idle(zones) + + +def main(): + # pylint: disable=too-many-locals,too-many-branches,too-many-statements + '''Receiver main loop. See README "code.py size constraint" for + why the lint disables above are required - exceeding ~32.5KB + breaks _bleio init on QT Py S3 with CP 10.1.4 (bisected). + ''' + print("MagicBand+ BLE Beacon Ears") + print(f"Rendering on {'onboard NeoPixel' if _USE_ONBOARD else 'stereo Jewels'}") + print(f"BLE scan: window={_SCAN_WINDOW_S * 1000:.0f}ms" + f" interval={_SCAN_INTERVAL_S * 1000:.0f}ms") + print("-" * 60) + + if _USE_ONBOARD: + pixels = pixel_zones.OnboardSingle(brightness=0.2) + else: + pixels = pixel_zones.StereoJewels( + left_pin=board.A1, right_pin=board.A3, + brightness=_BRIGHTNESS_PRESETS[0]) + zones = pixels.make_zones() + + adapter = _bleio.adapter + if not adapter.enabled: + adapter.enabled = True + + batt = battery_mod.BatteryMonitor() + + button_pin = digitalio.DigitalInOut(board.BUTTON) + button_pin.switch_to_input(pull=digitalio.Pull.UP) + button = Button(button_pin, value_when_pressed=False, + short_duration_ms=_BUTTON_SHORT_MS, + long_duration_ms=_BUTTON_LONG_MS) + + state = _RuntimeState() + + while True: + frame_start = time.monotonic() + + # --- BLE scan (still inline; Stages E and F will refactor) --- + new_command = None + try: + for entry in adapter.start_scan( + interval=_SCAN_INTERVAL_S, window=_SCAN_WINDOW_S, + minimum_rssi=_MIN_RSSI, + timeout=_SCAN_WINDOW_S, + extended=False, active=False): + payload = _extract_disney_payload(entry.advertisement_bytes) + if payload is None: + continue + if not payload: + continue + if _is_remote_brightness_trigger(payload): + now = time.monotonic() + if now - state.last_trigger_time < _TRIGGER_COOLDOWN_S: + continue + state.last_trigger_time = now + state.brightness_idx[0] = ( + (state.brightness_idx[0] + 1) % len(_BRIGHTNESS_PRESETS)) + new_b = _BRIGHTNESS_PRESETS[state.brightness_idx[0]] + pixels.set_brightness(new_b) + state.brightness_flash_until = now + _BRIGHTNESS_FLASH_DURATION_S + state.brightness_flash_level = state.brightness_idx[0] + continue + if _is_remote_battery_trigger(payload): + now = time.monotonic() + if now - state.last_trigger_time < _TRIGGER_COOLDOWN_S: + continue + state.last_trigger_time = now + batt.update(force=True) + state.battery_display_voltage = batt.voltage + v_str = (f"{state.battery_display_voltage:.3f}V" + if state.battery_display_voltage is not None else "None") + if (state.battery_display_voltage is not None + and state.battery_display_voltage > _USB_PRESENT_V_RAW): + print(f"[battery trigger remote] {v_str} (USB - showing display)") + state.battery_display_until = now + _BATTERY_DISPLAY_DURATION_S + state.battery_display_started_at = now + else: + print(f"[battery trigger remote] {v_str} (battery - yellow flash)") + state.unavailable_flash_until = now + _UNAVAILABLE_FLASH_DURATION_S + state.unavailable_flash_started_at = now + continue + if _is_remote_find_trigger(payload): + now = time.monotonic() + if now - state.last_trigger_time < _TRIGGER_COOLDOWN_S: + continue + state.last_trigger_time = now + print("[find me] starting high-visibility animation") + pixels.set_brightness(1.0) + state.find_mode_until = now + renderer.FIND_MODE_DURATION_S + state.find_mode_started_at = now + continue + if _is_remote_statue_trigger(payload): + now = time.monotonic() + if now - state.last_trigger_time < _TRIGGER_COOLDOWN_S: + continue + state.last_trigger_time = now + print("[statue preview] firing golden swirl") + fake_statue_cmd = { + "kind": "statue_beacon", + "statue_id": "PV", + "raw": bytes(payload), + } + candidate = renderer.for_command(fake_statue_cmd) + if candidate is not None: + new_command = (candidate, entry.rssi, bytes(payload)) + continue + is_mb = payload[0] in (0xE1, 0xE2, 0xCC) + is_wand = magicband_protocol.is_wand_packet(payload) + is_statue = (payload[0] == 0xC4 and len(payload) in (18, 23)) + if not (is_mb or is_wand or is_statue): + continue + now = time.monotonic() + if is_statue: + if now - state.last_statue_trigger < _STATUE_COOLDOWN_S: + continue + state.last_statue_trigger = now + else: + if (payload == state.last_payload + and now - state.last_payload_at < _DEDUP_WINDOW_S): + continue + state.last_payload = payload + state.last_payload_at = now + parsed = magicband_protocol.parse(payload) + candidate = renderer.for_command(parsed) + if candidate is not None: + new_command = (candidate, entry.rssi, payload) + finally: + adapter.stop_scan() + + # --- Helpers extracted in Stage D --- + _handle_button(state, button, batt, pixels) + + if new_command is not None: + state.active_state, rssi, raw = new_command + state.active_started_at = time.monotonic() + _log_command(state.active_state["label"], rssi, raw) + + if state.active_state is not None: + duration = state.active_state["duration_s"] + if duration is not None: + if time.monotonic() - state.active_started_at >= duration: + state.active_state = None + + _check_battery_log(state, batt) + _render_frame(zones, state, pixels) + pixels.show() + + elapsed = time.monotonic() - frame_start + remaining = _FRAME_BUDGET_S - elapsed + if remaining > 0: + time.sleep(remaining) + + +main() diff --git a/BLE_Beacon_Ears/magicband_protocol.py b/BLE_Beacon_Ears/magicband_protocol.py new file mode 100644 index 000000000..08e259534 --- /dev/null +++ b/BLE_Beacon_Ears/magicband_protocol.py @@ -0,0 +1,373 @@ +# SPDX-FileCopyrightText: 2026 Pedro Ruiz for Adafruit Industries +# +# SPDX-License-Identifier: MIT +'''MagicBand+ BLE protocol constants and helpers. + +Shared between the CLUE transmitter and the QT Py S3 receiver. Based on the +reverse-engineering work at: + https://emcot.world/Disney_MagicBand%2B_Bluetooth_Codes + +All command payloads stored here are the manufacturer-data portion only (the +bytes after the 0x0183 Disney CID). The transmitter prepends the CID bytes +when building a BLE advertisement packet. +''' +# Target: shared between the Adafruit CLUE (BLE remote) and the Adafruit +# QT Py ESP32-S3 (BLE Beacon Ears) - copy this file to both boards. + +# Disney's Bluetooth SIG company identifier. +DISNEY_CID = 0x0183 + +# 5-bit color palette. Values are RGB approximations calibrated for how the +# colors look on a NeoPixel Jewel at low brightness (~0.05). Green and blue +# channels look brighter per unit input than red on WS2812B LEDs, so cyan +# values have their red channel boosted to compensate, and blue hues get +# pushed toward their characteristic hue rather than a balanced RGB. +PALETTE_RGB = ( + (80, 255, 255), # 0x00 cyan (red channel boosted so it's not pure teal) + (180, 0, 255), # 0x01 purple + (0, 0, 255), # 0x02 blue + (0, 20, 120), # 0x03 midnight blue (touch of green stops it looking black) + (40, 120, 255), # 0x04 blue 2 + (200, 80, 255), # 0x05 bright purple + (200, 180, 255), # 0x06 lavender + (120, 0, 255), # 0x07 deep purple + (255, 60, 180), # 0x08 pink + (255, 70, 170), # 0x09 pink 2 + (255, 80, 160), # 0x0A pink 3 + (255, 90, 150), # 0x0B pink 4 + (255, 110, 150), # 0x0C pink 5 + (255, 130, 160), # 0x0D pink 6 + (255, 160, 170), # 0x0E pink 7 + (255, 180, 0), # 0x0F yellow orange + (255, 220, 0), # 0x10 off yellow + (255, 140, 20), # 0x11 yellow orange 2 + (180, 255, 0), # 0x12 lime + (255, 90, 0), # 0x13 orange + (255, 40, 0), # 0x14 red orange + (255, 0, 0), # 0x15 red + (60, 255, 255), # 0x16 cyan 2 (red boost for distinctness from green) + (40, 240, 255), # 0x17 cyan 3 + (20, 200, 255), # 0x18 cyan 4 (shifts more toward blue) + (0, 255, 0), # 0x19 green + (80, 255, 40), # 0x1A lime green + (255, 200, 180), # 0x1B white (warm white avoids blue cast at low levels) + (255, 200, 180), # 0x1C white 2 + (0, 0, 0), # 0x1D off + (255, 140, 60), # 0x1E unique + (255, 0, 255), # 0x1F random / magenta +) + +PALETTE_NAMES = ( + "Cyan", "Purple", "Blue", "Midnight Blue", + "Blue 2", "Bright Purple", "Lavender", "Deep Purple", + "Pink", "Pink 2", "Pink 3", "Pink 4", + "Pink 5", "Pink 6", "Pink 7", "Yellow Orange", + "Off Yellow", "Yellow Orange 2", "Lime", "Orange", + "Red Orange", "Red", "Cyan 2", "Cyan 3", + "Cyan 4", "Green", "Lime Green", "White", + "White 2", "Off", "Unique", "Random", +) + +# Mask palette: which of the 5 LEDs light up for a given 3-bit mask. +# Tuple order: (center, top_left, top_right, bottom_left, bottom_right) +MASK_LEDS = { + 0b000: (1, 1, 1, 1, 1), + 0b001: (0, 0, 1, 0, 0), + 0b010: (0, 0, 0, 0, 1), + 0b011: (0, 0, 0, 1, 0), + 0b100: (0, 1, 0, 0, 0), + 0b101: (1, 1, 1, 1, 1), + 0b110: (0, 0, 1, 0, 0), + 0b111: (1, 1, 1, 1, 1), +} + + +def decode_timing(byte): + '''Turn the timing byte into a dict of animation parameters.''' + scaler_b = bool(byte & 0x40) + time_val = byte & 0x0F + if scaler_b: + seconds = 3.1 * time_val + 5.5 + else: + seconds = 1.5 * time_val + 6.5 + return { + "always_on": bool(byte & 0x80), + "fade_code": (byte >> 4) & 0x03, + "seconds": seconds, + } + + +def build_single_color(palette_idx, mask=0, vibration=0, timing=0x09): + '''Build an E9 05 single-color-from-palette command payload.''' + color_byte = ((mask & 0x07) << 5) | (palette_idx & 0x1F) + vib_byte = 0xB0 | (vibration & 0x0F) + return bytes((0xE1, 0x00, 0xE9, 0x05, 0x00, timing, 0x0E, + color_byte, vib_byte)) + + +def build_dual_color(inner_idx, outer_idx, vibration=0, timing=0x22): + '''Build an E9 06 dual-color command payload. + + Note: the emcot wiki spec text says the top 3 bits of each color byte + should be 0b100, but the wiki's own example payloads use 0b010. Using + 0b100 causes the top-left LED to be masked off (same bits the E9 05 + mask palette uses for "top left only"). The correct working value + per the captured examples is 0b010 / 0x40. + ''' + inner_byte = 0x40 | (inner_idx & 0x1F) + outer_byte = 0x40 | (outer_idx & 0x1F) + vib_byte = 0xB0 | (vibration & 0x0F) + return bytes((0xE2, 0x00, 0xE9, 0x06, 0x00, timing, 0x0F, + inner_byte, outer_byte, vib_byte)) + + +def build_six_bit_color(red, green, blue, vibration=0, timing=0x0E): + '''Build an E9 08 raw 6-bit RGB command payload.''' + red_byte = (red & 0x3F) << 1 + green_byte = (green & 0x3F) << 1 + blue_byte = (blue & 0x3F) << 1 + vib_byte = 0xB0 | (vibration & 0x0F) + return bytes((0xE1, 0x00, 0xE9, 0x08, 0x00, timing, 0xD2, 0x55, + red_byte, green_byte, blue_byte, vib_byte)) + + +def build_five_color(center, top_left, bottom_left, bottom_right, top_right, + vibration=0, timing=0x0E): + '''Build an E9 09 five-color-palette command payload. + + Each of the band's 5 LEDs gets its own palette slot. Order matches the + emcot wiki byte order: center, bottom-left, bottom-right, top-right, + top-left (reading outer ring counter-clockwise from top-left). + ''' + def _color_byte(idx): + return 0xA0 | (idx & 0x1F) + vib_byte = 0xB0 | (vibration & 0x0F) + return bytes((0xE1, 0x00, 0xE9, 0x09, 0x00, timing, 0x0F, + _color_byte(top_left), + _color_byte(bottom_left), + _color_byte(bottom_right), + _color_byte(top_right), + _color_byte(center), + vib_byte)) + + +# Starlight Bubble Wand BLE protocol (reverse-engineered April 2026). +# 13-byte packets. First 6 bytes are a fixed signature identifying the +# wand and the "cast color" command. Bytes 6-11 contain a rolling code +# (probably anti-replay authentication) that changes on every broadcast. +# Byte 12 is the palette index - same table as the MagicBand+ palette. +# +# We only check the first 6 bytes to recognize a wand packet. The rolling +# middle bytes cannot be replayed (they would fail the wand's own checks +# if sent back), so we read them but don't try to decode or broadcast +# them ourselves. +WAND_SIGNATURE = bytes.fromhex("cf0b00c42022") +WAND_PAYLOAD_LENGTH = 13 +WAND_COLOR_INDEX = 12 + + +def is_wand_packet(payload): + '''Return True if this payload is a Starlight Bubble Wand cast.''' + return (len(payload) == WAND_PAYLOAD_LENGTH + and bytes(payload[:len(WAND_SIGNATURE)]) == WAND_SIGNATURE) + + +def parse_wand(payload): + '''Decode a wand cast packet into a structured command dict.''' + if not is_wand_packet(payload): + return None + palette_idx = payload[WAND_COLOR_INDEX] & 0x1F + return { + "kind": "wand_cast", + "palette_idx": palette_idx, + "raw": bytes(payload), + } + + +# Fab 50 statue beacons. The Disney Fab 50 golden statues placed around +# Magic Kingdom broadcast 0xC4 packets to assist guest location services. +# Two sub-formats: C4 10 (18 bytes) and C4 15 (22 bytes). Both contain +# an ASCII 2-digit statue ID at offset 15-16 (e.g. "53", "40", "24"). +# Triggering a golden-swirl animation when these are detected gives the +# wearable a thematic "the statue sees you" reaction. +_STATUE_PREFIX = bytes.fromhex("c4") + + +def _is_statue_beacon(payload): + '''Return True if this payload looks like a Fab 50 statue beacon.''' + if not payload or payload[0] != 0xC4: + return False + # Two known formats: C4 10 (18 bytes) and C4 15 (23 bytes) + return len(payload) in (18, 23) + + +def _parse_statue_beacon(payload): + '''Decode a statue beacon to extract its 2-digit ASCII identifier.''' + statue_id = "?" + # Statue ID is at offset 15-16 in both 18- and 22-byte variants + if len(payload) >= 17: + try: + statue_id = bytes(payload[15:17]).decode("ascii") + except (UnicodeError, ValueError): + statue_id = "?" + return { + "kind": "statue_beacon", + "statue_id": statue_id, + "raw": bytes(payload), + } + + +# Park show command opcodes - direct E9/EA family with no E1 00 wrapper. +# Captured from Disney park show infrastructure (Epcot, April 2026). These +# coexist with guest-fired E1/E2 commands but use a different byte layout. +# Long-format variants (E9 10, E9 13, EA 14) share a `f4 48 82` signature +# in the middle of the payload; their byte structure isn't fully decoded +# yet. The E9 08 short form decodes cleanly as a 5-slot palette command. +_SHOW_OPCODE_LABELS = { + (0xE9, 0x04): "E9 04", + (0xE9, 0x08): "E9 08 5-slot", + (0xE9, 0x10): "E9 10", + (0xE9, 0x13): "E9 13", + (0xEA, 0x14): "EA 14", +} + + +def _parse_show_command(payload): + '''Parse a direct E9/EA show packet captured from park infrastructure.''' + if len(payload) < 2: + return None + head = payload[0] + sub = payload[1] + label = _SHOW_OPCODE_LABELS.get((head, sub)) + if label is None: + return None + # E9 08 short form is a 5-slot palette command. Bytes 5-9 are masked + # with 0x1F to extract palette indices, identical to the existing E9 + # 09 five-color decode. Confirmed by capture 9 (blue green) decoding + # to Cyan/Blue 2/Green/Green/Blue 2 - matching the observed color. + slots = None + if (head == 0xE9 and sub == 0x08 + and len(payload) >= 10 and payload[4] == 0x0F): + slots = [payload[5 + i] & 0x1F for i in range(5)] + return { + "kind": "show_command", + "label": label, + "head": head, + "sub": sub, + "slots": slots, + "raw": bytes(payload), + } + + +def _parse_by_head(payload): + '''Decode a payload that's not a wand cast or statue beacon.''' + head = payload[0] + if head == 0xCC: + return {"kind": "ping", "raw": payload} + if head in (0xE9, 0xEA): + show_cmd = _parse_show_command(payload) + if show_cmd is not None: + return show_cmd + if head in (0xE1, 0xE2): + return _parse_e1_e2(payload) + return {"kind": "unknown", "raw": payload} + + +def parse(payload): + '''Decode a manufacturer-data payload into a structured command dict. + + Used by the QT Py receiver to interpret commands from MagicBands, the + CLUE remote, the Starlight Bubble Wand, and Disney park infrastructure + (Fab 50 statues, parade beacons). + ''' + if not payload: + return None + # Wand packets have a distinctive 6-byte header signature + wand = parse_wand(payload) + if wand is not None: + return wand + # Fab 50 statue beacons (Magic Kingdom hub area) + if _is_statue_beacon(payload): + return _parse_statue_beacon(payload) + return _parse_by_head(payload) + + +def _parse_single_color(payload): + color_byte = payload[7] + return { + "kind": "single_color", + "mask": (color_byte >> 5) & 0x07, + "palette_idx": color_byte & 0x1F, + "timing": decode_timing(payload[5]), + "vibration": payload[8] & 0x0F, + } + + +def _parse_dual_color(payload): + return { + "kind": "dual_color", + "inner_idx": payload[7] & 0x1F, + "outer_idx": payload[8] & 0x1F, + "timing": decode_timing(payload[5]), + "vibration": payload[9] & 0x0F, + } + + +def _parse_six_bit(payload): + return { + "kind": "six_bit_color", + "red": (payload[8] >> 1) & 0x3F, + "green": (payload[9] >> 1) & 0x3F, + "blue": (payload[10] >> 1) & 0x3F, + "timing": decode_timing(payload[5]), + "vibration": payload[11] & 0x0F, + } + + +def _parse_five_color(payload): + '''E9 09 layout: TL BL BR TR C VIB starting at index 7.''' + return { + "kind": "five_color", + "top_left": payload[7] & 0x1F, + "bottom_left": payload[8] & 0x1F, + "bottom_right": payload[9] & 0x1F, + "top_right": payload[10] & 0x1F, + "center": payload[11] & 0x1F, + "timing": decode_timing(payload[5]), + "vibration": payload[12] & 0x0F, + } + + +# Function-code dispatch for E1/E2-wrapped payloads. Each entry maps +# the 2-byte function code (payload[2]<<8 | payload[3]) to (min_length, +# parser_or_kind). When the parser slot is a callable, it's invoked with +# the payload; when it's a string, a generic {"kind": ..., "raw": ...} +# dict is returned. Defined at module bottom so all _parse_* helpers +# already exist when this dict is built at import time. +_FUNC_CODE_DISPATCH = { + 0xE905: (9, _parse_single_color), + 0xE906: (10, _parse_dual_color), + 0xE908: (12, _parse_six_bit), + 0xE909: (13, _parse_five_color), + 0xE90C: (5, "show_fx"), + 0xE911: (5, "cross_fade"), + # Newer parade/show command not in our protocol docs. We can't + # decode the colors but still want the ears to react visibly. + 0xCD07: (5, "parade_command"), +} + + +def _parse_e1_e2(payload): + '''Decode an E1/E2-wrapped payload by its function code.''' + if len(payload) < 5: + return {"kind": "unknown", "raw": payload} + func = (payload[2] << 8) | payload[3] + entry = _FUNC_CODE_DISPATCH.get(func) + if entry is None: + return {"kind": "animation", "func": func, "raw": payload} + min_len, handler = entry + if len(payload) < min_len: + return {"kind": "animation", "func": func, "raw": payload} + if callable(handler): + return handler(payload) + return {"kind": handler, "raw": payload} diff --git a/BLE_Beacon_Ears/pixel_zones.py b/BLE_Beacon_Ears/pixel_zones.py new file mode 100644 index 000000000..a75fe0c83 --- /dev/null +++ b/BLE_Beacon_Ears/pixel_zones.py @@ -0,0 +1,161 @@ +# SPDX-FileCopyrightText: 2026 Pedro Ruiz for Adafruit Industries +# +# SPDX-License-Identifier: MIT +'''Pixel zone abstraction for the BLE Beacon Ears project. + +A "zone" represents one ear. In prototype mode (ONBOARD) both zones composite +into the single onboard NeoPixel, with the left zone painting first and the +right zone painting second so the right side "wins" - still useful for a +quick visual of current state. In production mode (STEREO_JEWELS) each zone +drives its own 7-pixel Jewel on an independent data pin. + +The API is intentionally minimal: + zone.fill(rgb) - solid color across all pixels + zone.set_led(idx, rgb) - write a specific pixel + zone.count - number of pixels in the zone + zone.show() - flush to hardware (no-op if auto_write) + +The renderer writes to both zones each frame and calls show() once at frame +end. Double-buffering is not needed at 15fps - a torn frame would be +invisible to the eye. +''' +# Target: Adafruit QT Py ESP32-S3 - the BLE Beacon Ears +import board +import digitalio +import neopixel + + +class OnboardSingle: + '''Prototype mode: both zones share the single onboard NeoPixel. + + The left zone paints first; the right zone overwrites. This gives a + "right-biased" preview of stereo effects. Pair with the serial + stereo_log helper in the renderer to see what the other side is doing. + ''' + + def __init__(self, brightness=0.15): + self._power = None + try: + self._power = digitalio.DigitalInOut(board.NEOPIXEL_POWER) + self._power.switch_to_output(value=True) + except (AttributeError, ValueError): + pass + self._pixel = neopixel.NeoPixel( + board.NEOPIXEL, 1, brightness=brightness, auto_write=False) + self._pixel.fill((0, 0, 0)) + self._pixel.show() + + def make_zones(self): + '''Return (left_zone, right_zone) both pointing at the same pixel.''' + # Both zones share the underlying neopixel object. The renderer + # writes left first, then right. A single show() call at frame end + # flushes the right zone's final state to hardware. + return _SingleZone(self._pixel), _SingleZone(self._pixel) + + def show(self): + '''Flush the onboard pixel state to hardware.''' + self._pixel.show() + + def set_brightness(self, brightness): + '''Change pixel brightness at runtime.''' + self._pixel.brightness = brightness + + +class StereoJewels: + '''Production mode: two 7-pixel Jewels on independent data pins. + + Supports an idle-skip optimization: once both jewels have been shown + as all-black, subsequent show() calls are no-ops until pixel data + actually changes. This avoids unnecessary data-stream activity and + lets the WS2812 chips stay in their lowest-current latched state. + ''' + + def __init__(self, left_pin, right_pin, brightness=0.1): + self._left = neopixel.NeoPixel( + left_pin, 7, brightness=brightness, auto_write=False) + self._right = neopixel.NeoPixel( + right_pin, 7, brightness=brightness, auto_write=False) + self._left.fill((0, 0, 0)) + self._right.fill((0, 0, 0)) + self._left.show() + self._right.show() + self._last_shown_black = True + + def make_zones(self): + '''Return (left_zone, right_zone) wrapping each Jewel separately.''' + return _JewelZone(self._left), _JewelZone(self._right) + + def set_brightness(self, brightness): + '''Change the brightness of both jewels at runtime.''' + self._left.brightness = brightness + self._right.brightness = brightness + self._last_shown_black = False # force next show() to push new values + + def _all_black(self): + '''Return True if every pixel on both jewels is currently (0,0,0).''' + for i in range(7): + if self._left[i] != (0, 0, 0): + return False + if self._right[i] != (0, 0, 0): + return False + return True + + def show(self): + '''Flush both Jewel buffers to hardware, with idle-skip optimization. + + If we've already shown all-black once and nothing has changed to + non-black since, skip the data stream to save power. The WS2812 + chips latch their last color state and stay in quiescent mode + until new data arrives. + ''' + if self._all_black(): + if self._last_shown_black: + return + self._left.show() + self._right.show() + self._last_shown_black = True + else: + self._left.show() + self._right.show() + self._last_shown_black = False + + +class _SingleZone: + '''Zone backed by a shared 1-pixel NeoPixel (OnboardSingle mode).''' + + count = 1 + + def __init__(self, pixel): + self._pixel = pixel + + def fill(self, rgb): + '''Set the shared onboard pixel to the given color.''' + self._pixel[0] = rgb + + def set_led(self, idx, rgb): + '''Write a specific pixel index. Only idx 0 has effect here. + + All LED writes collapse to the single onboard pixel. The last + write of the frame wins, which matches the "right overwrites + left" compositing intent. + ''' + if idx == 0: + self._pixel[0] = rgb + + +class _JewelZone: + '''Zone backed by a dedicated 7-pixel Jewel (StereoJewels mode).''' + + count = 7 + + def __init__(self, pixel_obj): + self._pixel = pixel_obj + + def fill(self, rgb): + '''Set every pixel on this Jewel to the given color.''' + self._pixel.fill(rgb) + + def set_led(self, idx, rgb): + '''Write a specific pixel index (0-6) on this Jewel.''' + if 0 <= idx < 7: + self._pixel[idx] = rgb diff --git a/BLE_Beacon_Ears/renderer.py b/BLE_Beacon_Ears/renderer.py new file mode 100644 index 000000000..98667dff3 --- /dev/null +++ b/BLE_Beacon_Ears/renderer.py @@ -0,0 +1,839 @@ +# SPDX-FileCopyrightText: 2026 Pedro Ruiz for Adafruit Industries +# +# SPDX-License-Identifier: MIT +'''Animation renderer for MagicBand+ commands. + +Given a parsed command dict from magicband_protocol, produces an +AnimationState that the game loop can render over time. The state is a +small dict holding: + started_at (seconds) - time.monotonic() when command received + duration_s (float or None) - how long until animation stops; None = forever + render(zones, t) - callback that paints a frame at time t + +Design rules: + - Renderer NEVER blocks. All timing is derived from t (time since start). + - All palette colors are looked up via magicband_protocol.PALETTE_RGB. + - Stereo effects are applied at render time via per-zone phase offset. + - When no active animation exists, zones are filled black. +''' +# Target: Adafruit QT Py ESP32-S3 - the BLE Beacon Ears +import math + +import magicband_protocol + +# Breathing envelope - applied to static colors to add life. +# Amplitude 0.45 means brightness oscillates between 55% and 100% of target. +# This is more pronounced than a subtle 25% amplitude would be - on 7-pixel +# jewels with full color, you need more contrast to read as "breathing." +_BREATH_PERIOD_S = 2.5 +_BREATH_AMPLITUDE = 0.45 + +# Rotation period for animations without explicit timing cues. +_ROTATE_PERIOD_S = 2.0 + +# Default duration when a command has no parseable timing byte. +_DEFAULT_DURATION_S = 10.0 + +# Cross-fade period for dual-color alternation on pixel prototype. +_DUAL_ALTERNATE_PERIOD_S = 1.2 + + +def _scale_rgb(rgb, factor): + '''Multiply each channel by factor (0.0-1.0) and clamp.''' + return ( + min(255, max(0, int(rgb[0] * factor))), + min(255, max(0, int(rgb[1] * factor))), + min(255, max(0, int(rgb[2] * factor))), + ) + + +def _breath_factor(t, phase=0.0): + '''Return 0..1 brightness multiplier that breathes gently over time.''' + # Sinusoid, output range (1 - amp) to 1.0. + wave = 0.5 + 0.5 * math.sin( + 2 * math.pi * (t / _BREATH_PERIOD_S + phase)) + return (1.0 - _BREATH_AMPLITUDE) + _BREATH_AMPLITUDE * wave + + +def for_command(command): + '''Build an AnimationState from a parsed command dict.''' + kind = command.get("kind", "unknown") + handler = _COMMAND_HANDLERS.get(kind) + if handler is None: + return None + if kind == "ping": + return handler() + return handler(command) + + +def _state_parade_command(_command): + '''CD 07 / parade beacon - colors not decoded. + + Newer Disney park show commands (Starlight Parade etc.) use formats + we haven't reverse-engineered. Rather than ignoring them and being + silent during the show, render a generic rainbow rotation so the + ears at least respond visibly. The wearer sees that the show IS + triggering them, just not in the exact same way as a real band. + ''' + rainbow_colors = ( + (255, 0, 0), # red + (255, 100, 0), # orange + (255, 220, 0), # yellow + (0, 255, 0), # green + (0, 120, 255), # blue + (180, 0, 255), # purple + ) + + def render(zones, t): + # Smooth rotation over 2 seconds per cycle + phase = t / 2.0 + n_colors = len(rainbow_colors) + for zone_idx, zone in enumerate(zones): + zone_offset = 0.5 if zone_idx == 1 else 0.0 + outer_count = max(1, zone.count - 1) + # Center pixel cycles slowly + center_slot = int(phase) % n_colors + zone.set_led(0, _scale_rgb(rainbow_colors[center_slot], 0.4)) + for led_idx in range(1, zone.count): + angle_frac = (led_idx - 1) / outer_count + color_phase = phase + zone_offset + angle_frac + slot = int(color_phase * n_colors) % n_colors + zone.set_led(led_idx, rainbow_colors[slot]) + + return {"duration_s": 5.0, "render": render, "label": "PARADE"} + + +def _state_statue_beacon(command): + '''Fab 50 statue detected - golden swirl with sparkles and pulse. + + Disney's Fab 50 golden statues at Magic Kingdom broadcast continuous + location beacons. We trigger this animation to acknowledge "the + statue sees you" when one is detected nearby. The original 2-second + version felt too brief - now 4.0s with two pulse "beats" so the + surprise lasts long enough to register. + + Visual: warm gold pixels swirl around the outer ring, with random + bright white sparkles popping in. Sparkles get easier to trigger + on pulse peaks. A breathing brightness envelope creates two beats + over the 4-second span; each beat peaks mid-rotation, so the swirl + waxes bright at ~1s and ~3s with a softer dip between them. + ''' + statue_id = command.get("statue_id", "?") + gold_bright = (255, 180, 30) + gold_dim = (120, 80, 10) + sparkle_white = (255, 255, 200) + + def _statue_pixel_for(led_idx, ring_idx, zone_idx, t, phase, envelope, pulse): + '''Compute the RGB for one outer-ring LED in the statue swirl.''' + # Sparkle: pseudo-random per LED + time. Threshold gets + # slightly easier on pulse peaks so sparkles cluster + # rhythmically with the beat instead of feeling random. + sparkle_phase = t * 12.0 + led_idx * 1.7 + zone_idx * 0.5 + if math.sin(sparkle_phase) > 0.88 - 0.06 * pulse: + return _scale_rgb(sparkle_white, envelope) + zone_offset = phase + (0.3 if zone_idx == 1 else 0.0) + outer_count = 6 + head_pos = (zone_offset * outer_count) % outer_count + distance = (head_pos - ring_idx) % outer_count + if distance < 1.0: + return _scale_rgb(gold_bright, envelope) + if distance < 3.0: + fade = 1.0 - (distance / 3.0) + return _scale_rgb(gold_dim, envelope * fade) + return (0, 0, 0) + + def render(zones, t): + # Outer envelope: 0-0.3s fade in, 3.7-4.0s fade out, flat between. + if t < 0.3: + fade_envelope = t / 0.3 + elif t > 3.7: + fade_envelope = max(0.0, (4.0 - t) / 0.3) + else: + fade_envelope = 1.0 + # Two-beat pulse: dim at t=0, peak at t=1, dim at t=2, peak at + # t=3, dim at t=4. Cosine-shifted sine keeps range tight (0.5..1.0) + # so the swirl never disappears entirely between peaks. + pulse = 0.75 + 0.25 * math.sin( + 2 * math.pi * t / 2.0 - math.pi / 2) + envelope = fade_envelope * pulse + # Continuous rotation - no reset between beats. ~3 full + # revolutions over the 4s span at 1.5 rev/s base rate. + phase = t * 1.5 + for zone_idx, zone in enumerate(zones): + # Center pixel: steady warm gold modulated by envelope + zone.set_led(0, (int(gold_bright[0] * envelope * 0.6), + int(gold_bright[1] * envelope * 0.6), + int(gold_bright[2] * envelope * 0.6))) + for led_idx in range(1, zone.count): + zone.set_led(led_idx, _statue_pixel_for( + led_idx, led_idx - 1, zone_idx, t, phase, + envelope, pulse)) + + return { + "duration_s": 4.0, + "render": render, + "label": f"STATUE #{statue_id}", + } + + +# Find Me beacon - 3-phase high-visibility animation for locating a +# stroller, wheelchair, or EV scooter in a busy parking lot. Triggered +# by the CLUE remote's "Find Me" command. The main loop also forces +# the pixel brightness to maximum during this animation regardless of +# the user's preset, then restores their preset after. +_FIND_STROBE_S = 3.0 # Phase 1: attention-grabbing strobe +_FIND_CHASE_S = 15.0 # Phase 2: rainbow chase (motion + color) +_FIND_BREATHE_S = 12.0 # Phase 3: rainbow breathing (steady glow) +FIND_MODE_DURATION_S = _FIND_STROBE_S + _FIND_CHASE_S + _FIND_BREATHE_S + + +def _state_find_me(_command): + '''3-phase high-visibility "find me" animation. + + Phase 1: Strobe - rapid full-white + saturated color flashes to + catch eyes from across a parking lot. + Phase 2: Rainbow chase - bright pixels rotating around the ring with + rainbow color trail. Easy to spot at distance, indicates motion + / liveness. + Phase 3: Rainbow breathing - steady saturated rainbow at slower + breath rate. Less alarming once located, easy to home in on. + ''' + # Vivid colors used throughout (full saturation for visibility) + rainbow = ( + (255, 0, 0), # red + (255, 80, 0), # orange + (255, 200, 0), # yellow + (0, 255, 0), # green + (0, 80, 255), # blue + (180, 0, 255), # purple + ) + white = (255, 255, 255) + + def _phase_strobe(zones, t): + '''Phase 1: strobe at 5 Hz alternating white and rainbow color.''' + strobe_idx = int(t * 10) # 10 strobes/sec + if strobe_idx % 2 == 0: + color = white + else: + color = rainbow[(strobe_idx // 2) % len(rainbow)] + for zone in zones: + zone.fill(color) + + def _phase_chase(zones, phase_t): + '''Phase 2: rainbow chase rotating around the ring.''' + rotation = phase_t * 1.5 # ~1.5 revolutions per second + n = len(rainbow) + for zone_idx, zone in enumerate(zones): + zone_offset = rotation + (0.3 if zone_idx == 1 else 0.0) + zone.set_led(0, rainbow[int(rotation * 0.5) % n]) + outer_count = max(1, zone.count - 1) + for led_idx in range(1, zone.count): + angle = (led_idx - 1) / outer_count + zone.set_led(led_idx, + rainbow[int((zone_offset + angle) * n) % n]) + + def _phase_breathe(zones, phase_t): + '''Phase 3: rainbow breathing - all pixels rainbow with envelope.''' + breath = 0.5 + 0.5 * math.sin(2 * math.pi * phase_t / 2.0) + n = len(rainbow) + for zone_idx, zone in enumerate(zones): + zone_offset = 0.5 if zone_idx == 1 else 0.0 + zone.set_led(0, _scale_rgb( + rainbow[int(phase_t / 2.0) % n], breath)) + outer_count = max(1, zone.count - 1) + for led_idx in range(1, zone.count): + angle = (led_idx - 1) / outer_count + zone.set_led(led_idx, _scale_rgb( + rainbow[int((angle + zone_offset) * n) % n], breath)) + + def render(zones, t): + if t < _FIND_STROBE_S: + _phase_strobe(zones, t) + elif t < _FIND_STROBE_S + _FIND_CHASE_S: + _phase_chase(zones, t - _FIND_STROBE_S) + else: + _phase_breathe(zones, t - _FIND_STROBE_S - _FIND_CHASE_S) + + return { + "duration_s": FIND_MODE_DURATION_S, + "render": render, + "label": "FIND ME", + } + + +def _lighten(rgb, amount=0.4): + '''Return rgb shifted toward white by `amount` (0.0-1.0).''' + return ( + min(255, int(rgb[0] + (255 - rgb[0]) * amount)), + min(255, int(rgb[1] + (255 - rgb[1]) * amount)), + min(255, int(rgb[2] + (255 - rgb[2]) * amount)), + ) + + +def _state_wand_cast(command): + '''Starlight Wand cast - a multi-phase comet animation. + + Narrative: + 1. Comet swirls on the LEFT ear twice (outer ring), tail fading behind. + 2. Comet "crosses over" briefly (both ears show a quick trail). + 3. Comet swirls on the RIGHT ear twice. + 4. Both ears sparkle with a lighter shade of the cast color. + 5. Both ears settle into a slow breathing glow of the cast color, + holding for 30 seconds or until the next command arrives. + + Each outer-ring pixel on a 7-pixel Jewel is indexed 1..6 with pixel 0 + being the center. The comet moves around the 6 outer pixels; the + center pixel glows at a modest fraction to anchor the swirl. + ''' + palette_idx = command["palette_idx"] + rgb = magicband_protocol.PALETTE_RGB[palette_idx] + sparkle_rgb = _lighten(rgb, 0.55) + name = magicband_protocol.PALETTE_NAMES[palette_idx] + + # Phase timings (seconds from t=0): swirl_left_end, crossover_end, + # swirl_right_end, sparkle_end, total_duration. + timings = (0.9, 1.1, 2.0, 2.8, 30.0) + # Comet shape: outer_count, tail_len, tail_falloff (per-step dimming). + comet_shape = (6, 4, 0.55) + + def _comet_on_zone(zone, head_position, color_bright): + '''Draw a comet with tail on the outer ring of one zone. + + head_position is a float 0..outer_count-1 indicating where the + comet "head" sits on the ring. Tail trails behind it at decreasing + brightness. Center pixel (idx 0) anchors at a modest glow. + ''' + outer_count, tail_len, tail_falloff = comet_shape + # Fade the center so it's subtle but present + zone.set_led(0, _scale_rgb(color_bright, 0.3)) + # Tail effect - for each outer ring pixel, compute its distance + # back from the head and set brightness accordingly. + head_int = int(head_position) % outer_count + for ring_idx in range(outer_count): + led_idx = ring_idx + 1 # skip center + # Distance back from head (always positive, wrapping around) + distance = (head_int - ring_idx) % outer_count + if distance > tail_len: + zone.set_led(led_idx, (0, 0, 0)) + elif distance == 0: + # Bright head + zone.set_led(led_idx, color_bright) + else: + factor = tail_falloff ** distance + zone.set_led(led_idx, _scale_rgb(color_bright, factor)) + + def _dark_zone(zone): + for i in range(zone.count): + zone.set_led(i, (0, 0, 0)) + + def render(zones, t): + left, right = zones[0], zones[1] + swirl_left_end, crossover_end, swirl_right_end, sparkle_end, _ = timings + outer_count = comet_shape[0] + + if t < swirl_left_end: + _wand_phase_left(left, right, t, swirl_left_end, + outer_count, rgb, _comet_on_zone, _dark_zone) + elif t < crossover_end: + _wand_phase_crossover(left, right, t, swirl_left_end, + crossover_end, outer_count, rgb, + _comet_on_zone) + elif t < swirl_right_end: + _wand_phase_right(left, right, t, crossover_end, + swirl_right_end, outer_count, rgb, + _comet_on_zone, _dark_zone) + elif t < sparkle_end: + _wand_phase_sparkle(zones, t, rgb, sparkle_rgb) + else: + _wand_phase_breathe(left, right, t - sparkle_end, rgb) + + return { + "duration_s": timings[4], + "render": render, + "label": f"WAND {name}", + } + + +def _wand_phase_left(left, right, t, swirl_left_end, outer_count, + rgb, comet_fn, dark_fn): + '''Phase 1: left ear comet, 2 full swirls. Right ear dark.''' + progress = t / swirl_left_end + comet_fn(left, progress * outer_count * 2, rgb) + dark_fn(right) + + +def _wand_phase_crossover(left, right, t, swirl_left_end, crossover_end, + outer_count, rgb, comet_fn): + '''Phase 2: crossover - left fades, right starts.''' + fade_progress = (t - swirl_left_end) / (crossover_end - swirl_left_end) + comet_fn(left, outer_count * 2 - 1, + _scale_rgb(rgb, 1.0 - fade_progress)) + comet_fn(right, 0, _scale_rgb(rgb, fade_progress)) + + +def _wand_phase_right(left, right, t, crossover_end, swirl_right_end, + outer_count, rgb, comet_fn, dark_fn): + '''Phase 3: right ear comet, 2 full swirls. Left ear dark.''' + progress = (t - crossover_end) / (swirl_right_end - crossover_end) + dark_fn(left) + comet_fn(right, progress * outer_count * 2, rgb) + + +def _wand_phase_sparkle(zones, t, rgb, sparkle_rgb): + '''Phase 4: sparkle burst - both ears scatter light shimmers.''' + for zone_idx, zone in enumerate(zones): + for led_idx in range(zone.count): + phase = t * 22.0 + led_idx * 1.3 + zone_idx * 0.7 + twinkle = abs(math.sin(phase * 2 * math.pi)) + if twinkle > 0.75: + zone.set_led(led_idx, sparkle_rgb) + elif twinkle > 0.4: + zone.set_led(led_idx, _scale_rgb(rgb, twinkle)) + else: + zone.set_led(led_idx, _scale_rgb(rgb, 0.2)) + + +def _wand_phase_breathe(left, right, t_breath, rgb): + '''Phase 5: settled breathing - slow, gentle pulse on both ears.''' + # 4-second period, 30% amplitude (between 70% and 100%). Right ear + # is offset by half a cycle so the two ears breathe out of phase. + left_f = 0.85 + 0.15 * math.sin(2 * math.pi * t_breath / 4.0) + right_f = 0.85 + 0.15 * math.sin( + 2 * math.pi * (t_breath / 4.0 + 0.5)) + left.fill(_scale_rgb(rgb, left_f)) + right.fill(_scale_rgb(rgb, right_f)) + + +def _state_ping(): + # Wake-ping packets (CC03) are fired by the CLUE remote right before + # commands flagged needs_ping=True. They're a meta-signal meant for + # the band receiver, not something the ears should visualize. Return + # None so the game loop ignores it entirely. + return None + + +def _state_single_color(command): + palette = command["palette_idx"] + rgb = magicband_protocol.PALETTE_RGB[palette] + name = magicband_protocol.PALETTE_NAMES[palette] + duration = None if command["timing"]["always_on"] else command["timing"]["seconds"] + + def render(zones, t): + # Stereo breathing: left and right out of phase by half a cycle. + left_rgb = _scale_rgb(rgb, _breath_factor(t, phase=0.0)) + right_rgb = _scale_rgb(rgb, _breath_factor(t, phase=0.5)) + zones[0].fill(left_rgb) + zones[1].fill(right_rgb) + + return { + "duration_s": duration, + "render": render, + "label": f"SINGLE {name}", + } + + +def _state_dual_color(command): + inner = magicband_protocol.PALETTE_RGB[command["inner_idx"]] + outer = magicband_protocol.PALETTE_RGB[command["outer_idx"]] + inner_name = magicband_protocol.PALETTE_NAMES[command["inner_idx"]] + outer_name = magicband_protocol.PALETTE_NAMES[command["outer_idx"]] + duration = None if command["timing"]["always_on"] else command["timing"]["seconds"] + + def render(zones, t): + # Stereo assignment: left = inner, right = outer. Each still breathes + # to stay lively. Out-of-phase breathing keeps the two ears feeling + # alive rather than identical. + zones[0].fill(_scale_rgb(inner, _breath_factor(t, phase=0.0))) + zones[1].fill(_scale_rgb(outer, _breath_factor(t, phase=0.5))) + + return { + "duration_s": duration, + "render": render, + "label": f"DUAL {inner_name}/{outer_name}", + } + + +def _state_five_color(command): + # Each of the 5 band LEDs has its own palette slot. We pick a + # representative color per zone: left zone uses top-left + bottom-left, + # right zone uses top-right + bottom-right, each zone's center lights up + # the average. For single-pixel prototype we flatten further. + tl = magicband_protocol.PALETTE_RGB[command["top_left"]] + bl = magicband_protocol.PALETTE_RGB[command["bottom_left"]] + tr = magicband_protocol.PALETTE_RGB[command["top_right"]] + br = magicband_protocol.PALETTE_RGB[command["bottom_right"]] + center = magicband_protocol.PALETTE_RGB[command["center"]] + duration = (None if command["timing"]["always_on"] + else command["timing"]["seconds"]) + + def _avg(a, b, c): + return ((a[0] + b[0] + c[0]) // 3, + (a[1] + b[1] + c[1]) // 3, + (a[2] + b[2] + c[2]) // 3) + + left_rgb = _avg(tl, bl, center) + right_rgb = _avg(tr, br, center) + + def render(zones, t): + zones[0].fill(_scale_rgb(left_rgb, _breath_factor(t, phase=0.0))) + zones[1].fill(_scale_rgb(right_rgb, _breath_factor(t, phase=0.5))) + + return {"duration_s": duration, "render": render, "label": "FIVE"} + + +def _state_six_bit(command): + # E9 08 gives raw 6-bit RGB. Expand to 8-bit and use it directly. + rgb = (command["red"] << 2, command["green"] << 2, command["blue"] << 2) + duration = (None if command["timing"]["always_on"] + else command["timing"]["seconds"]) + + def render(zones, t): + zones[0].fill(_scale_rgb(rgb, _breath_factor(t, phase=0.0))) + zones[1].fill(_scale_rgb(rgb, _breath_factor(t, phase=0.5))) + + return {"duration_s": duration, "render": render, "label": "RGB6"} + + +# Firmware-baked E9 0C animations. These have known payload signatures +# whose bytes are NOT raw 5-slot palette indices but rather animation +# program selectors. We map them to approximate visual color sequences +# matching how real MagicBand+ hardware actually plays them. +# +# Key: first 12 bytes of payload (signature prefix, excludes timing/vib/vib) +# Value: (label, ordered color RGB list) +_BAKED_ANIMATIONS = { + # Taste the Rainbow - full rainbow rotation + bytes.fromhex("e100e90c000f0f5d465bf005"): ( + "Rainbow", + [ + (255, 0, 0), # red + (255, 90, 0), # orange + (255, 220, 0), # yellow + (0, 255, 0), # green + (0, 120, 255), # blue + (180, 0, 255), # purple + ], + ), + # Blink White - white strobe + bytes.fromhex("e100e90c000f0f5d465bf005"): ( + "Blink White", + [(255, 220, 200), (0, 0, 0)], + ), + # Orange Blink - orange pulse + bytes.fromhex("e100e90c00ef0f4f4f5bf0fb"): ( + "Orange Blink", + [(255, 90, 0), (50, 20, 0)], + ), +} + + +def _lookup_baked_animation(raw): + '''Return (label, slots) if raw matches a known firmware animation.''' + # Taste the Rainbow and Blink White share the same 12-byte prefix but + # differ in tail bytes. Disambiguate by comparing tail too. + prefix = bytes(raw[:12]) + tail = bytes(raw[12:]) if len(raw) > 12 else b"" + + # Taste the Rainbow full: e100e90c000f0f5d465bf005 32 37 48 b0 + # Blink White full: e100e90c000f0f5d465bf005 32 37 48 95 + # Distinguished by last byte: b0=no vibration (TTR), 95=other (Blink White) + if prefix == bytes.fromhex("e100e90c000f0f5d465bf005"): + if len(tail) >= 4 and tail[-1] == 0x95: + return ("Blink White", + [(255, 220, 200), (0, 0, 0)]) + # Default this prefix to Taste the Rainbow + return ("Rainbow", + [(255, 0, 0), (255, 90, 0), (255, 220, 0), + (0, 255, 0), (0, 120, 255), (180, 0, 255)]) + + # Orange Blink + if prefix == bytes.fromhex("e100e90c00ef0f4f4f5bf0fb"): + return ("Orange Blink", + [(255, 90, 0), (50, 20, 0)]) + + return None + + +def _decode_5slot_palette(raw): + '''Decode bytes 7..11 of a 5-slot E9 0C payload into colors and label. + + Returns (slots, label) where slots is a list of RGB tuples and label + is a short summary of the distinct color names involved. + ''' + slot_bytes = raw[7:12] if len(raw) >= 12 else raw[7:] + slots = [] + slot_names = [] + for byte in slot_bytes: + idx = byte & 0x1F + slots.append(magicband_protocol.PALETTE_RGB[idx]) + slot_names.append(magicband_protocol.PALETTE_NAMES[idx]) + if not slots: + slots = [(255, 255, 255)] + slot_names = ["White"] + distinct = [] + for name in slot_names: + short = name.split()[0][:4] + if not distinct or distinct[-1] != short: + distinct.append(short) + return slots, f"SHOW {'>'.join(distinct)}" + + +def _state_show_fx(command): + '''E9 0C captured park animations. + + Some E9 0C payloads are firmware-baked animation programs (Taste the + Rainbow, Blink White, Orange Blink) where the bytes are program IDs, + not 5-slot palettes. We recognize those by signature and use hardcoded + color sequences matching their real visual appearance. + + Other E9 0C payloads (5 Palette Cycle, DCL Rainbow, future clones) are + true 5-slot palette cycles - for those we extract colors from bytes + 7-11 as palette indices. + ''' + raw = command["raw"] + baked = _lookup_baked_animation(raw) + if baked is not None: + label_suffix, slots = baked + label = f"SHOW {label_suffix}" + else: + slots, label = _decode_5slot_palette(raw) + + duration = _DEFAULT_DURATION_S + if len(raw) >= 6: + timing = magicband_protocol.decode_timing(raw[5]) + duration = None if timing["always_on"] else timing["seconds"] + + def render(zones, t): + n_slots = len(slots) + phase = t / _ROTATE_PERIOD_S + for zone_idx, zone in enumerate(zones): + zone_offset = 0.5 if zone_idx == 1 else 0.0 + center_slot = int(phase) % n_slots + zone.set_led(0, slots[center_slot]) + outer_count = max(1, zone.count - 1) + for led_idx in range(1, zone.count): + angle_frac = (led_idx - 1) / outer_count + color_phase = phase + zone_offset + angle_frac + slot = int(color_phase * n_slots) % n_slots + zone.set_led(led_idx, slots[slot]) + + return {"duration_s": duration, "render": render, "label": label} + + +def _state_cross_fade(command): + # E9 11 cross fade between two palette colors. The two endpoint colors + # are encoded in bytes 7 (from) and 8 (to) of the payload. Bytes 9+ + # appear to be fade timing and repeat parameters, not additional color + # slots. + raw = command["raw"] + slot_a_idx = (raw[7] if len(raw) > 7 else 0) & 0x1F + slot_b_idx = (raw[8] if len(raw) > 8 else 0) & 0x1F + slot_a = magicband_protocol.PALETTE_RGB[slot_a_idx] + slot_b = magicband_protocol.PALETTE_RGB[slot_b_idx] + name_a = magicband_protocol.PALETTE_NAMES[slot_a_idx] + name_b = magicband_protocol.PALETTE_NAMES[slot_b_idx] + duration = _DEFAULT_DURATION_S * 2 + if len(raw) >= 6: + timing = magicband_protocol.decode_timing(raw[5]) + duration = None if timing["always_on"] else timing["seconds"] + + def _mix(a, b, f): + return ( + int(a[0] * (1 - f) + b[0] * f), + int(a[1] * (1 - f) + b[1] * f), + int(a[2] * (1 - f) + b[2] * f), + ) + + def render(zones, t): + # Slow sinusoidal cross fade between a and b. Left leads right. + f_left = 0.5 + 0.5 * math.sin(2 * math.pi * t / 4.0) + f_right = 0.5 + 0.5 * math.sin(2 * math.pi * (t - 1.0) / 4.0) + zones[0].fill(_mix(slot_a, slot_b, f_left)) + zones[1].fill(_mix(slot_a, slot_b, f_right)) + + return { + "duration_s": duration, + "render": render, + "label": f"FADE {name_a}<>{name_b}", + } + + +def _show_command_slots_path(slots): + '''5-slot palette path for show_command. Returns a state dict.''' + slot_colors = [magicband_protocol.PALETTE_RGB[i] for i in slots] + slot_names = [magicband_protocol.PALETTE_NAMES[i] for i in slots] + distinct = [] + for name in slot_names: + short = name.split()[0][:4] + if not distinct or distinct[-1] != short: + distinct.append(short) + full_label = f"SHOW5 {'>'.join(distinct)}" + + def render_slots(zones, t): + n_slots = len(slot_colors) + phase = t / _ROTATE_PERIOD_S + for zone_idx, zone in enumerate(zones): + zone_offset = 0.5 if zone_idx == 1 else 0.0 + zone.set_led(0, slot_colors[int(phase) % n_slots]) + outer_count = max(1, zone.count - 1) + for led_idx in range(1, zone.count): + angle_frac = (led_idx - 1) / outer_count + color_phase = phase + zone_offset + angle_frac + zone.set_led(led_idx, + slot_colors[int(color_phase * n_slots) % n_slots]) + + return {"duration_s": _DEFAULT_DURATION_S, "render": render_slots, + "label": full_label} + + +def _show_command_generic_path(raw, label): + '''Generic park-show pulse for un-decoded long-format packets. + + Uses a position-weighted polynomial hash to derive a deterministic + primary palette index per capture - simple XOR collapsed multiple + captures into the same bucket, which defeats the "tell captures + apart on camera" goal. + ''' + seed = 0 + for byte in raw: + seed = (seed * 31 + byte) & 0xFFFF + palette_size = len(magicband_protocol.PALETTE_RGB) + primary_idx = seed % palette_size + # Skip the "Off" palette entry so the primary is never black. + if magicband_protocol.PALETTE_RGB[primary_idx] == (0, 0, 0): + primary_idx = (primary_idx + 1) % palette_size + primary_name = magicband_protocol.PALETTE_NAMES[primary_idx] + # Anchor the primary plus three accents spaced around the palette. + accents = ( + magicband_protocol.PALETTE_RGB[primary_idx], + magicband_protocol.PALETTE_RGB[(primary_idx + 6) % palette_size], + magicband_protocol.PALETTE_RGB[(primary_idx + 12) % palette_size], + magicband_protocol.PALETTE_RGB[(primary_idx + 18) % palette_size], + ) + + def render_generic(zones, t): + # 1.2s rotation - faster than the 2.0s used for guest commands, + # gives the show pulse a more energetic feel. + phase = t / 1.2 + n_slots = 4 + for zone_idx, zone in enumerate(zones): + zone_offset = 0.5 if zone_idx == 1 else 0.0 + zone.set_led(0, accents[int(phase) % n_slots]) + outer_count = max(1, zone.count - 1) + for led_idx in range(1, zone.count): + angle_frac = (led_idx - 1) / outer_count + color_phase = phase + zone_offset + angle_frac + zone.set_led(led_idx, + accents[int(color_phase * n_slots) % n_slots]) + + return {"duration_s": _DEFAULT_DURATION_S, "render": render_generic, + "label": f"{label} hue={primary_name}"} + + +def _state_show_command(command): + '''Park-show packet renderer (Epcot light show, etc.). + + Two paths depending on whether the payload decodes: + - If `slots` is set (E9 08 short form), render as a 5-slot palette + rotation matching firmware show_fx output. + - Otherwise the long-format payloads (E9 10, E9 13, EA 14) aren't + fully decoded yet, so render a generic park-show pulse with a + primary hue derived from the payload bytes. Different captured + payloads produce visibly different primary colors, so multiple + captures can be told apart on camera even though we can't decode + their internal structure. + ''' + slots = command.get("slots") + if slots is not None: + return _show_command_slots_path(slots) + return _show_command_generic_path( + command["raw"], command.get("label", "SHOW")) + + +def _state_animation(command): + # Generic animation (E9 0B, E9 0E, E9 0F, etc.). The jewels have 7 + # pixels each (1 center + 6 outer ring), so we can do real spatial + # rotation: cycle the outer ring through the palette slots while + # keeping the center a fixed color. Reads as a proper "color wheel" + # effect like the real bands. + raw = command["raw"] + slots = [] + slot_names = [] + # Most animation payloads have color bytes after the 7-byte header. + # Skip the last byte (vibration) when collecting colors. + for byte in raw[7:-1]: + idx = byte & 0x1F + if idx < 0x1F: # skip obvious non-color bytes like vibration codes + slots.append(magicband_protocol.PALETTE_RGB[idx]) + slot_names.append(magicband_protocol.PALETTE_NAMES[idx]) + if not slots: + slots = [(100, 100, 100)] + slot_names = ["Gray"] + func = command.get("func", 0) + + # Label lists distinct color short names for at-a-glance recognition. + distinct = [] + for name in slot_names: + short = name.split()[0][:4] + if not distinct or distinct[-1] != short: + distinct.append(short) + # Cap label length so serial output stays readable + label_colors = '>'.join(distinct[:4]) + if len(distinct) > 4: + label_colors += '...' + label = f"ANIM 0x{func:04X} {label_colors}" + + def render(zones, t): + n_slots = len(slots) + # How fast the color wheel rotates - one full revolution per period + rotations_per_s = 1.0 / _ROTATE_PERIOD_S + # Global phase advances linearly with time + phase = t * rotations_per_s + for zone_idx, zone in enumerate(zones): + zone_offset = 0.5 if zone_idx == 1 else 0.0 # right trails left + # Center LED gets the "middle" slot as an anchor color + center_slot = int(phase) % n_slots + zone.set_led(0, slots[center_slot]) + # Outer ring LEDs (1-6) each get a color at their angular + # position, offset by the rotating phase. count=7 means + # indices 1..6 are outer pixels for the NeoPixel Jewel. + outer_count = max(1, zone.count - 1) + for led_idx in range(1, zone.count): + # Each outer pixel's color index walks around the palette + # based on its physical angular position + the rotation phase + angle_frac = (led_idx - 1) / outer_count + color_phase = phase + zone_offset + angle_frac + slot = int(color_phase * n_slots) % n_slots + zone.set_led(led_idx, slots[slot]) + + return { + "duration_s": _DEFAULT_DURATION_S, + "render": render, + "label": label, + } + + +# Dispatch table for for_command(). Defined after the _state_* functions +# so all references resolve. Handlers all take a command dict, except +# _state_ping which takes no args. +_COMMAND_HANDLERS = { + "ping": _state_ping, + "wand_cast": _state_wand_cast, + "single_color": _state_single_color, + "dual_color": _state_dual_color, + "five_color": _state_five_color, + "six_bit_color": _state_six_bit, + "show_fx": _state_show_fx, + "cross_fade": _state_cross_fade, + "animation": _state_animation, + "parade_command": _state_parade_command, + "show_command": _state_show_command, + "statue_beacon": _state_statue_beacon, + "find_me": _state_find_me, +} + + +def render_idle(zones): + '''Default renderer when no animation is active. Blanks both zones.''' + zones[0].fill((0, 0, 0)) + zones[1].fill((0, 0, 0)) From 154fac9937b8368659c399a54f0a3ad6247b47b2 Mon Sep 17 00:00:00 2001 From: videopixil Date: Thu, 7 May 2026 18:36:40 -0400 Subject: [PATCH 2/3] Update code.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Removed unused import neopixel line 32339 bytes (16 bytes smaller, slightly more headroom under the cliff) Lint clean even with unused-import enforced (Adafruit CI doesn't disable that one — my local config did, mistake on my part) Behavior unchanged — neopixel was never actually used here, only inside pixel_zones.py --- BLE_Beacon_Ears/code.py | 1 - 1 file changed, 1 deletion(-) diff --git a/BLE_Beacon_Ears/code.py b/BLE_Beacon_Ears/code.py index b3673f2d0..9ef97dbf9 100755 --- a/BLE_Beacon_Ears/code.py +++ b/BLE_Beacon_Ears/code.py @@ -19,7 +19,6 @@ import _bleio import board import digitalio -import neopixel from adafruit_debouncer import Button import battery as battery_mod From aeb27792ed6b0d0557ebc6d3f2e8b76aeca46fae Mon Sep 17 00:00:00 2001 From: videopixil Date: Fri, 8 May 2026 20:48:36 -0400 Subject: [PATCH 3/3] fixes to code, main and added battery.py applied fixes to code, removed main and added battery.py --- BLE_Beacon_Ears/battery.py | 67 ++++ BLE_Beacon_Ears/code.py | 680 ++++++++++++++++----------------- BLE_Beacon_Ears/pixel_zones.py | 69 +--- 3 files changed, 389 insertions(+), 427 deletions(-) create mode 100644 BLE_Beacon_Ears/battery.py diff --git a/BLE_Beacon_Ears/battery.py b/BLE_Beacon_Ears/battery.py new file mode 100644 index 000000000..b5ae97fbf --- /dev/null +++ b/BLE_Beacon_Ears/battery.py @@ -0,0 +1,67 @@ +# SPDX-FileCopyrightText: 2026 Pedro Ruiz for Adafruit Industries +# +# SPDX-License-Identifier: MIT +'''Simple battery voltage monitor for the Beacon Ears. + +Reads the BFF's voltage divider on board.A2 and exposes a `voltage` +property. We keep this minimal: no state machine, no auto-warnings, no +ambient rendering. The voltage is consumed only by the on-demand +battery display in code.py when the user explicitly asks for it. + +Note: the ESP32-S3 ADC reads voltages higher than actual (uncalibrated +ADC gain error). We don't try to correct this - instead the thresholds +in code.py are tuned to match what the QT Py actually reports for +known battery states. +''' +# Target: Adafruit QT Py ESP32-S3 - the BLE Beacon Ears +import time + +import analogio +import board + +_ADC_FULL_SCALE = 65535 +_DIVIDER_RATIO = 2.0 + +_SAMPLES_PER_READ = 10 +_SAMPLE_INTERVAL_S = 5.0 + + +class BatteryMonitor: + '''Polls battery voltage on A2. No state machine, no auto-rendering.''' + + def __init__(self): + try: + self._pin = analogio.AnalogIn(board.A2) + except (AttributeError, ValueError): + self._pin = None + self._last_read_t = 0.0 + self._last_voltage = None + + def _read_voltage(self): + '''Read and average N ADC samples, return raw volts on the cell.''' + if self._pin is None: + return None + total = 0 + for _ in range(_SAMPLES_PER_READ): + total += self._pin.value + avg = total / _SAMPLES_PER_READ + ref_v = self._pin.reference_voltage + return (avg / _ADC_FULL_SCALE) * ref_v * _DIVIDER_RATIO + + def update(self, force=False): + '''Refresh voltage reading if enough time has passed. + + If force=True, read voltage immediately regardless of interval. + Useful when responding to a user trigger - we want the value + shown to reflect the moment of the press, not 5 seconds ago. + ''' + now = time.monotonic() + if not force and now - self._last_read_t < _SAMPLE_INTERVAL_S: + return + self._last_read_t = now + self._last_voltage = self._read_voltage() + + @property + def voltage(self): + '''Last measured raw voltage in volts (None if unread/absent).''' + return self._last_voltage diff --git a/BLE_Beacon_Ears/code.py b/BLE_Beacon_Ears/code.py index 9ef97dbf9..ea6533d04 100755 --- a/BLE_Beacon_Ears/code.py +++ b/BLE_Beacon_Ears/code.py @@ -7,10 +7,12 @@ NeoPixel Jewels. Two logical zones (left ear, right ear) are rendered with stereo phase offsets to give static colors a gentle out-of-phase breathing animation and rotations a left-leads-right sweep. - -To switch from real Jewels to onboard NeoPixel for testing without the -ear hardware, flip _USE_ONBOARD to True below. ''' +# pylint: disable=redefined-outer-name +# Setup and main loop are inlined at module level (no main() wrapper). +# Helper functions defined above take parameters named the same as the +# inlined module-level loop variables (zone, t, etc) - this is expected +# and harmless: helpers and the inline body never share scope at runtime. # Target: Adafruit QT Py ESP32-S3 - the BLE Beacon Ears import math import random @@ -26,9 +28,6 @@ import pixel_zones import renderer -# Flip this to True to test on the QT Py's onboard pixel without jewels. -_USE_ONBOARD = False - # 15 fps target -> ~66ms per frame. _TARGET_FPS = 15 _FRAME_BUDGET_S = 1.0 / _TARGET_FPS @@ -66,7 +65,7 @@ # How long the BOOT button must be held to count as a long press. # Short press cycles brightness preset (the more frequently-used action). # Long press shows the battery level (less frequent, requires intent). -_LONG_PRESS_S = 0.5 +LONG_PRESS_S = 0.5 _BRIGHTNESS_FLASH_DURATION_S = 0.6 # USB-presence detection threshold (raw QT Py voltage reading). @@ -146,8 +145,8 @@ # committing a multi-press count. 350ms is a common sweet spot - tight # enough that single-press brightness cycling still feels responsive, # wide enough that comfortable triple-presses register reliably. -_BUTTON_SHORT_MS = 350 -_BUTTON_LONG_MS = int(_LONG_PRESS_S * 1000) +BUTTON_SHORT_MS = 350 +BUTTON_LONG_MS = int(LONG_PRESS_S * 1000) def _build_showpieces(): @@ -374,26 +373,21 @@ def _render_battery_display(zones, voltage, t): # 0x03 = cycle brightness preset # 0x04 = find me # 0x05 = statue animation preview -_REMOTE_BATTERY_PACKET = bytes.fromhex("aa4201") -_REMOTE_BRIGHTNESS_PACKET = bytes.fromhex("aa4203") -_REMOTE_FIND_PACKET = bytes.fromhex("aa4204") -_REMOTE_STATUE_PACKET = bytes.fromhex("aa4205") - - -def _is_remote_battery_trigger(payload): - return bytes(payload[:3]) == _REMOTE_BATTERY_PACKET - - -def _is_remote_brightness_trigger(payload): - return bytes(payload[:3]) == _REMOTE_BRIGHTNESS_PACKET - +# Remote-trigger packet table. Sub-protocol: AA42xx is sent only by the +# CLUE remote. Real bands and wands ignore packets they don't recognize, +# so this is safe alongside MagicBand+ (E1/E2/CC), wand (CF), and +# Fab 50 statue (C4) traffic. +REMOTE_COMMANDS = { + bytes.fromhex("aa4201"): "battery", + bytes.fromhex("aa4203"): "brightness", + bytes.fromhex("aa4204"): "find", + bytes.fromhex("aa4205"): "statue", +} -def _is_remote_find_trigger(payload): - return bytes(payload[:3]) == _REMOTE_FIND_PACKET - -def _is_remote_statue_trigger(payload): - return bytes(payload[:3]) == _REMOTE_STATUE_PACKET +def remote_command(payload): + '''Return the remote-trigger command name, or None if not a trigger.''' + return REMOTE_COMMANDS.get(bytes(payload[:3])) def _extract_disney_payload(ad_bytes): @@ -418,374 +412,338 @@ def _log_command(label, rssi, raw): print(f" raw={raw.hex()}") - -class _RuntimeState: - '''All mutable state for the receiver main loop in one object. - - Bundling state into a single object keeps main() under the lint limit - for local variables and makes the per-frame helpers cleanly testable. - Plain data holder, no methods, like a struct. - ''' - # pylint: disable=too-many-instance-attributes - def __init__(self): - # Active rendered animation - self.active_state = None - self.active_started_at = 0.0 - # Dedup tracking for incoming BLE packets - self.last_payload = None - self.last_payload_at = 0.0 - # Brightness flash UI feedback - self.brightness_flash_until = 0.0 - self.brightness_flash_level = 0 - self.brightness_idx = [0] # list-wrapped for mutation reasons - # Trigger cooldowns to suppress repeated remote commands - self.last_trigger_time = 0.0 - self.last_statue_trigger = 0.0 - # On-demand battery display state - self.battery_display_until = 0.0 - self.battery_display_started_at = 0.0 - self.battery_display_voltage = None - # Yellow "plug in" flash (battery display unavailable) - self.unavailable_flash_until = 0.0 - self.unavailable_flash_started_at = 0.0 - # Find Me beacon mode - self.find_mode_until = 0.0 - self.find_mode_started_at = 0.0 - # Periodic battery voltage logging - self.last_battery_log = 0.0 - # Solo mode (curated showpiece reel) - self.solo_mode = False - self.solo_state = None - self.solo_label = "" - self.solo_idx = -1 - self.solo_started_at = 0.0 - self.solo_indicator_until = 0.0 - self.solo_indicator_started_at = 0.0 - self.solo_indicator_kind = None # 'enter' or 'exit' - - +print("MagicBand+ BLE Beacon Ears") +print(f"BLE scan: window={_SCAN_WINDOW_S * 1000:.0f}ms" + f" interval={_SCAN_INTERVAL_S * 1000:.0f}ms") +print("-" * 60) + +pixels = pixel_zones.StereoJewels( + left_pin=board.A1, right_pin=board.A3, + brightness=_BRIGHTNESS_PRESETS[0]) +zones = pixels.make_zones() +brightness_idx = [0] + +adapter = _bleio.adapter +if not adapter.enabled: + adapter.enabled = True + +batt = battery_mod.BatteryMonitor() + +button_pin = digitalio.DigitalInOut(board.BUTTON) +button_pin.switch_to_input(pull=digitalio.Pull.UP) +button = Button(button_pin, value_when_pressed=False, + short_duration_ms=BUTTON_SHORT_MS, + long_duration_ms=BUTTON_LONG_MS) + +active_state = None +active_started_at = 0.0 +last_payload = None +last_payload_at = 0.0 +brightness_flash_until = 0.0 +brightness_flash_level = 0 +last_trigger_time = 0.0 +battery_display_until = 0.0 +battery_display_started_at = 0.0 +battery_display_voltage = None # voltage snapshot at trigger time +unavailable_flash_until = 0.0 # yellow flash for "plug in to check" +unavailable_flash_started_at = 0.0 +last_statue_trigger = 0.0 # last Fab 50 statue swirl fired +find_mode_until = 0.0 # find-me beacon animation end time +find_mode_started_at = 0.0 +last_battery_log = 0.0 _BATTERY_LOG_INTERVAL_S = 60.0 # log raw voltage once per minute - -def _handle_button(state, button, batt, pixels): - '''Process one tick of BOOT-button input. - - Single-press: brightness cycle. Double-press (in solo): skip - showpiece. Triple-press: toggle solo mode. Long-press: battery. - ''' +# Solo mode state. solo_state is a renderer animation dict (same shape +# as active_state); solo_label / solo_idx / solo_started_at track the +# current showpiece. solo_indicator_* drives the enter/exit pulse. +solo_mode = False +solo_state = None +solo_label = "" +solo_idx = -1 +solo_started_at = 0.0 +solo_indicator_until = 0.0 +solo_indicator_started_at = 0.0 +solo_indicator_kind = None # 'enter' or 'exit' + +while True: + frame_start = time.monotonic() + + new_command = None + try: + for entry in adapter.start_scan( + interval=_SCAN_INTERVAL_S, window=_SCAN_WINDOW_S, + minimum_rssi=_MIN_RSSI, + timeout=_SCAN_WINDOW_S, + extended=False, active=False): + payload = _extract_disney_payload(entry.advertisement_bytes) + if payload is None: + continue + if not payload: + continue + # Remote-trigger commands from the CLUE share a 4s cooldown. + command = remote_command(payload) + if command is not None: + now = time.monotonic() + if now - last_trigger_time < _TRIGGER_COOLDOWN_S: + continue + last_trigger_time = now + if command == "brightness": + brightness_idx[0] = ( + (brightness_idx[0] + 1) % len(_BRIGHTNESS_PRESETS)) + pixels.set_brightness(_BRIGHTNESS_PRESETS[brightness_idx[0]]) + brightness_flash_until = now + _BRIGHTNESS_FLASH_DURATION_S + brightness_flash_level = brightness_idx[0] + elif command == "battery": + # On battery power, WS2812 timing corruption makes + # the full animation unreliable (green renders as + # red). Show a brief yellow "plug in" flash on + # battery; full animation only on USB. + batt.update(force=True) + battery_display_voltage = batt.voltage + v_str = (f"{battery_display_voltage:.3f}V" + if battery_display_voltage is not None else "None") + if (battery_display_voltage is not None + and battery_display_voltage > _USB_PRESENT_V_RAW): + print(f"[battery trigger remote] {v_str} (USB)") + battery_display_until = now + _BATTERY_DISPLAY_DURATION_S + battery_display_started_at = now + else: + print(f"[battery trigger remote] {v_str} (yellow flash)") + unavailable_flash_until = now + _UNAVAILABLE_FLASH_DURATION_S + unavailable_flash_started_at = now + elif command == "find": + # Forces max brightness for the 30s high-visibility + # animation, then restores the user's preset. + print("[find me] starting high-visibility animation") + pixels.set_brightness(1.0) + find_mode_until = now + renderer.FIND_MODE_DURATION_S + find_mode_started_at = now + elif command == "statue": + # Fires the same golden swirl that real Fab 50 + # statue beacons trigger - useful for demos. + print("[statue preview] firing golden swirl") + candidate = renderer.for_command({ + "kind": "statue_beacon", + "statue_id": "PV", + "raw": bytes(payload), + }) + if candidate is not None: + new_command = (candidate, entry.rssi, bytes(payload)) + continue + # Accept MagicBand+ commands (E1/E2/CC), wand casts (CF), + # and Fab 50 statue beacons (C4). Statue beacons trigger + # a special golden-swirl animation rather than rendering + # any of their content. + is_mb = payload[0] in (0xE1, 0xE2, 0xCC) + is_wand = magicband_protocol.is_wand_packet(payload) + is_statue = (payload[0] == 0xC4 and len(payload) in (18, 23)) + if not (is_mb or is_wand or is_statue): + continue + now = time.monotonic() + # Statues broadcast many packets per second per statue. + # Use a longer cooldown specifically for statue triggers + # to avoid restarting the swirl animation constantly. + if is_statue: + if now - last_statue_trigger < _STATUE_COOLDOWN_S: + continue + last_statue_trigger = now + else: + if (payload == last_payload + and now - last_payload_at < _DEDUP_WINDOW_S): + continue + last_payload = payload + last_payload_at = now + parsed = magicband_protocol.parse(payload) + candidate = renderer.for_command(parsed) + if candidate is not None: + new_command = (candidate, entry.rssi, payload) + finally: + adapter.stop_scan() + + # --- Button check (BOOT button via adafruit_debouncer.Button) --- + # short_count=1 -> cycle brightness + # short_count=2 -> skip showpiece (solo mode only) + # short_count>=3 -> toggle solo mode + # long_press -> show battery (USB) or yellow "plug in" flash button.update() if button.long_press: batt.update(force=True) - state.battery_display_voltage = batt.voltage - v_str = (f"{state.battery_display_voltage:.3f}V" - if state.battery_display_voltage is not None else "None") + battery_display_voltage = batt.voltage + v_str = (f"{battery_display_voltage:.3f}V" + if battery_display_voltage is not None else "None") now = time.monotonic() - if (state.battery_display_voltage is not None - and state.battery_display_voltage > _USB_PRESENT_V_RAW): + if (battery_display_voltage is not None + and battery_display_voltage > _USB_PRESENT_V_RAW): print(f"[battery trigger BOOT] {v_str} (USB - showing display)") - state.battery_display_until = now + _BATTERY_DISPLAY_DURATION_S - state.battery_display_started_at = now + battery_display_until = now + _BATTERY_DISPLAY_DURATION_S + battery_display_started_at = now else: print(f"[battery trigger BOOT] {v_str} (battery - yellow flash)") - state.unavailable_flash_until = now + _UNAVAILABLE_FLASH_DURATION_S - state.unavailable_flash_started_at = now + unavailable_flash_until = now + _UNAVAILABLE_FLASH_DURATION_S + unavailable_flash_started_at = now elif button.short_count >= 3: # Triple-press toggles solo mode. now = time.monotonic() - state.solo_indicator_started_at = now - if state.solo_mode: - state.solo_mode = False - state.solo_state = None - state.solo_indicator_kind = "exit" - state.solo_indicator_until = now + _SOLO_INDICATOR_EXIT_S + solo_indicator_started_at = now + if solo_mode: + solo_mode = False + solo_state = None + solo_indicator_kind = "exit" + solo_indicator_until = now + _SOLO_INDICATOR_EXIT_S print("[solo] exiting") else: - state.solo_mode = True - state.solo_idx = -1 - state.solo_state = None - state.solo_indicator_kind = "enter" - state.solo_indicator_until = now + _SOLO_INDICATOR_ENTER_S + solo_mode = True + solo_idx = -1 # forces a fresh pick on next render + solo_state = None + solo_indicator_kind = "enter" + solo_indicator_until = now + _SOLO_INDICATOR_ENTER_S print("[solo] entering") - elif button.short_count == 2 and state.solo_mode: - state.solo_state = None + elif button.short_count == 2 and solo_mode: + # Double-press in solo: skip to a new random showpiece. + solo_state = None # next render block picks a new one print("[solo] skipping to next showpiece") elif button.short_count == 1: - state.brightness_idx[0] = ( - (state.brightness_idx[0] + 1) % len(_BRIGHTNESS_PRESETS)) - new_b = _BRIGHTNESS_PRESETS[state.brightness_idx[0]] + # Single press: cycle brightness preset. + brightness_idx[0] = (brightness_idx[0] + 1) % len(_BRIGHTNESS_PRESETS) + new_b = _BRIGHTNESS_PRESETS[brightness_idx[0]] pixels.set_brightness(new_b) - state.brightness_flash_until = ( - time.monotonic() + _BRIGHTNESS_FLASH_DURATION_S) - state.brightness_flash_level = state.brightness_idx[0] - - -def _check_battery_log(state, batt): - '''Throttled battery voltage logger - one line per minute.''' + brightness_flash_until = time.monotonic() + _BRIGHTNESS_FLASH_DURATION_S + brightness_flash_level = brightness_idx[0] + + # --- State update --- + if new_command is not None: + active_state, rssi, raw = new_command + active_started_at = time.monotonic() + _log_command(active_state["label"], rssi, raw) + + # --- Expiration check --- + if active_state is not None: + duration = active_state["duration_s"] + if duration is not None: + t = time.monotonic() - active_started_at + if t >= duration: + active_state = None + + # --- Battery monitor --- + # update() internally throttles to every 5 seconds, so calling + # every frame is cheap. No state machine - we just keep a fresh + # voltage available for the on-demand display. batt.update() + # Log raw voltage occasionally to help tune thresholds now = time.monotonic() if (batt.voltage is not None - and now - state.last_battery_log >= _BATTERY_LOG_INTERVAL_S): - state.last_battery_log = now + and now - last_battery_log >= _BATTERY_LOG_INTERVAL_S): + last_battery_log = now level = _battery_level_match(batt.voltage) level_str = f"{level[0]} pixels" if level else "CRITICAL" print(f"[battery] raw={batt.voltage:.3f}V -> {level_str}") - -def _render_brightness_flash(zones, state, frame_t): - '''Brightness-cycle confirmation flash on the outer ring.''' - flash_t = (_BRIGHTNESS_FLASH_DURATION_S - - (state.brightness_flash_until - frame_t)) - frac = flash_t / _BRIGHTNESS_FLASH_DURATION_S - if frac < 0.15: - envelope = frac / 0.15 - elif frac < 0.65: - envelope = 1.0 - else: - envelope = max(0.0, (1.0 - frac) / 0.35) - count = state.brightness_flash_level + 1 - lit = (int(255 * envelope),) * 3 - for zone in zones: - zone.set_led(0, (0, 0, 0)) - for i in range(1, zone.count): - zone.set_led(i, lit if (i - 1) < count else (0, 0, 0)) - - -def _render_unavailable_flash(zones, state, frame_t): - '''Yellow center pulse - "plug in to check battery".''' - flash_t = frame_t - state.unavailable_flash_started_at - frac = flash_t / _UNAVAILABLE_FLASH_DURATION_S - envelope = (frac * 2) if frac < 0.5 else max(0.0, 2 * (1 - frac)) - color = (int(_UNAVAILABLE_FLASH_COLOR[0] * envelope), - int(_UNAVAILABLE_FLASH_COLOR[1] * envelope), - int(_UNAVAILABLE_FLASH_COLOR[2] * envelope)) - for zone in zones: - zone.set_led(0, color) - for i in range(1, zone.count): - zone.set_led(i, (0, 0, 0)) - - -def _render_active_state(zones, state, frame_t): - '''Render the currently active animation, with crash safety.''' - t = frame_t - state.active_started_at - try: - state.active_state["render"](zones, t) - except Exception as err: # pylint: disable=broad-except - print(f"RENDER ERROR at t={t:.2f}s: " - f"{type(err).__name__}: {err}") - state.active_state = None - renderer.render_idle(zones) - - -def _render_solo_cycle(zones, state, frame_t): - '''Render one frame of solo mode\'s curated showpiece reel.''' - if state.solo_state is None: - state.solo_idx = _pick_showpiece_idx(state.solo_idx) - state.solo_label, parsed = _SHOWPIECES[state.solo_idx] - state.solo_state = renderer.for_command(parsed) - state.solo_started_at = frame_t - print(f"[solo] now playing: {state.solo_label}") - duration = state.solo_state["duration_s"] or _SOLO_DEFAULT_DURATION_S - t = frame_t - state.solo_started_at - if t >= duration + _SOLO_BREATH_S: - state.solo_state = None - renderer.render_idle(zones) - elif t >= duration: - renderer.render_idle(zones) - else: - try: - state.solo_state["render"](zones, t) - except Exception as err: # pylint: disable=broad-except - print(f"SOLO RENDER ERROR ({state.solo_label}) at " - f"t={t:.2f}s: {type(err).__name__}: {err}") - state.solo_state = None - renderer.render_idle(zones) - - -def _render_frame(zones, state, pixels): - '''Render one frame, picking the highest-priority active source.''' + # --- Render frame --- frame_t = time.monotonic() - if 0.0 < state.find_mode_until <= frame_t: - pixels.set_brightness(_BRIGHTNESS_PRESETS[state.brightness_idx[0]]) - state.find_mode_until = 0.0 + # Find Me beacon takes priority over EVERYTHING - including + # brightness flashes, animations, idle. Forces visibility for + # the full 30 seconds. When it ends, restore the user's + # brightness preset. + if 0.0 < find_mode_until <= frame_t: + # Animation just finished - restore user's brightness preset + pixels.set_brightness(_BRIGHTNESS_PRESETS[brightness_idx[0]]) + find_mode_until = 0.0 print("[find me] animation complete, brightness restored") - if frame_t < state.find_mode_until: - t = frame_t - state.find_mode_started_at + if frame_t < find_mode_until: + t = frame_t - find_mode_started_at + # Build a synthetic command dict to invoke the renderer find_state = renderer.for_command({"kind": "find_me"}) if find_state is not None: find_state["render"](zones, t) - elif frame_t < state.solo_indicator_until: - t = frame_t - state.solo_indicator_started_at - if state.solo_indicator_kind == "enter": + elif frame_t < solo_indicator_until: + # Solo enter (white burst) or exit (cool blue pulse). Sized + # to read clearly on camera for B-roll of the toggle moment. + t = frame_t - solo_indicator_started_at + if solo_indicator_kind == "enter": _render_solo_enter(zones, t) else: _render_solo_exit(zones, t) - elif frame_t < state.brightness_flash_until: - _render_brightness_flash(zones, state, frame_t) - elif frame_t < state.unavailable_flash_until: - _render_unavailable_flash(zones, state, frame_t) - elif frame_t < state.battery_display_until: - t = frame_t - state.battery_display_started_at - _render_battery_display(zones, state.battery_display_voltage, t) - elif state.active_state is not None: - _render_active_state(zones, state, frame_t) - elif state.solo_mode: - _render_solo_cycle(zones, state, frame_t) + elif frame_t < brightness_flash_until: + flash_t = _BRIGHTNESS_FLASH_DURATION_S - (brightness_flash_until - frame_t) + frac = flash_t / _BRIGHTNESS_FLASH_DURATION_S + if frac < 0.15: + envelope = frac / 0.15 + elif frac < 0.65: + envelope = 1.0 + else: + envelope = max(0.0, (1.0 - frac) / 0.35) + count = brightness_flash_level + 1 + lit = (int(255 * envelope),) * 3 + for zone in zones: + zone.set_led(0, (0, 0, 0)) + for i in range(1, zone.count): + ring_idx = i - 1 + if ring_idx < count: + zone.set_led(i, lit) + else: + zone.set_led(i, (0, 0, 0)) + elif frame_t < unavailable_flash_until: + # Brief yellow center pulse - "plug in to check battery" + # Yellow uses both R and G channels heavily, which keeps it + # readable even if WS2812 timing corrupts on battery. + flash_t = frame_t - unavailable_flash_started_at + frac = flash_t / _UNAVAILABLE_FLASH_DURATION_S + # Triangle envelope: ramp up first half, ramp down second + envelope = (frac * 2) if frac < 0.5 else max(0.0, 2 * (1 - frac)) + c = _UNAVAILABLE_FLASH_COLOR + color = (int(c[0] * envelope), + int(c[1] * envelope), + int(c[2] * envelope)) + for zone in zones: + zone.set_led(0, color) + for i in range(1, zone.count): + zone.set_led(i, (0, 0, 0)) + elif frame_t < battery_display_until: + t = frame_t - battery_display_started_at + _render_battery_display(zones, battery_display_voltage, t) + elif active_state is not None: + t = frame_t - active_started_at + try: + active_state["render"](zones, t) + except Exception as err: # pylint: disable=broad-except + print(f"RENDER ERROR at t={t:.2f}s: " + f"{type(err).__name__}: {err}") + active_state = None + renderer.render_idle(zones) + elif solo_mode: + # Cycle through curated showpieces. A real BLE packet sets + # active_state above this branch and preempts solo cleanly; + # solo resumes (with a fresh pick) once the BLE animation + # ends, so park interaction works without manual toggling. + if solo_state is None: + solo_idx = _pick_showpiece_idx(solo_idx) + solo_label, parsed = _SHOWPIECES[solo_idx] + solo_state = renderer.for_command(parsed) + solo_started_at = frame_t + print(f"[solo] now playing: {solo_label}") + duration = solo_state["duration_s"] or _SOLO_DEFAULT_DURATION_S + t = frame_t - solo_started_at + if t >= duration + _SOLO_BREATH_S: + # Showpiece + breath gap done; clear so next frame picks. + solo_state = None + renderer.render_idle(zones) + elif t >= duration: + # In the breath gap between showpieces. + renderer.render_idle(zones) + else: + try: + solo_state["render"](zones, t) + except Exception as err: # pylint: disable=broad-except + print(f"SOLO RENDER ERROR ({solo_label}) at " + f"t={t:.2f}s: {type(err).__name__}: {err}") + solo_state = None + renderer.render_idle(zones) else: renderer.render_idle(zones) + pixels.show() - -def main(): - # pylint: disable=too-many-locals,too-many-branches,too-many-statements - '''Receiver main loop. See README "code.py size constraint" for - why the lint disables above are required - exceeding ~32.5KB - breaks _bleio init on QT Py S3 with CP 10.1.4 (bisected). - ''' - print("MagicBand+ BLE Beacon Ears") - print(f"Rendering on {'onboard NeoPixel' if _USE_ONBOARD else 'stereo Jewels'}") - print(f"BLE scan: window={_SCAN_WINDOW_S * 1000:.0f}ms" - f" interval={_SCAN_INTERVAL_S * 1000:.0f}ms") - print("-" * 60) - - if _USE_ONBOARD: - pixels = pixel_zones.OnboardSingle(brightness=0.2) - else: - pixels = pixel_zones.StereoJewels( - left_pin=board.A1, right_pin=board.A3, - brightness=_BRIGHTNESS_PRESETS[0]) - zones = pixels.make_zones() - - adapter = _bleio.adapter - if not adapter.enabled: - adapter.enabled = True - - batt = battery_mod.BatteryMonitor() - - button_pin = digitalio.DigitalInOut(board.BUTTON) - button_pin.switch_to_input(pull=digitalio.Pull.UP) - button = Button(button_pin, value_when_pressed=False, - short_duration_ms=_BUTTON_SHORT_MS, - long_duration_ms=_BUTTON_LONG_MS) - - state = _RuntimeState() - - while True: - frame_start = time.monotonic() - - # --- BLE scan (still inline; Stages E and F will refactor) --- - new_command = None - try: - for entry in adapter.start_scan( - interval=_SCAN_INTERVAL_S, window=_SCAN_WINDOW_S, - minimum_rssi=_MIN_RSSI, - timeout=_SCAN_WINDOW_S, - extended=False, active=False): - payload = _extract_disney_payload(entry.advertisement_bytes) - if payload is None: - continue - if not payload: - continue - if _is_remote_brightness_trigger(payload): - now = time.monotonic() - if now - state.last_trigger_time < _TRIGGER_COOLDOWN_S: - continue - state.last_trigger_time = now - state.brightness_idx[0] = ( - (state.brightness_idx[0] + 1) % len(_BRIGHTNESS_PRESETS)) - new_b = _BRIGHTNESS_PRESETS[state.brightness_idx[0]] - pixels.set_brightness(new_b) - state.brightness_flash_until = now + _BRIGHTNESS_FLASH_DURATION_S - state.brightness_flash_level = state.brightness_idx[0] - continue - if _is_remote_battery_trigger(payload): - now = time.monotonic() - if now - state.last_trigger_time < _TRIGGER_COOLDOWN_S: - continue - state.last_trigger_time = now - batt.update(force=True) - state.battery_display_voltage = batt.voltage - v_str = (f"{state.battery_display_voltage:.3f}V" - if state.battery_display_voltage is not None else "None") - if (state.battery_display_voltage is not None - and state.battery_display_voltage > _USB_PRESENT_V_RAW): - print(f"[battery trigger remote] {v_str} (USB - showing display)") - state.battery_display_until = now + _BATTERY_DISPLAY_DURATION_S - state.battery_display_started_at = now - else: - print(f"[battery trigger remote] {v_str} (battery - yellow flash)") - state.unavailable_flash_until = now + _UNAVAILABLE_FLASH_DURATION_S - state.unavailable_flash_started_at = now - continue - if _is_remote_find_trigger(payload): - now = time.monotonic() - if now - state.last_trigger_time < _TRIGGER_COOLDOWN_S: - continue - state.last_trigger_time = now - print("[find me] starting high-visibility animation") - pixels.set_brightness(1.0) - state.find_mode_until = now + renderer.FIND_MODE_DURATION_S - state.find_mode_started_at = now - continue - if _is_remote_statue_trigger(payload): - now = time.monotonic() - if now - state.last_trigger_time < _TRIGGER_COOLDOWN_S: - continue - state.last_trigger_time = now - print("[statue preview] firing golden swirl") - fake_statue_cmd = { - "kind": "statue_beacon", - "statue_id": "PV", - "raw": bytes(payload), - } - candidate = renderer.for_command(fake_statue_cmd) - if candidate is not None: - new_command = (candidate, entry.rssi, bytes(payload)) - continue - is_mb = payload[0] in (0xE1, 0xE2, 0xCC) - is_wand = magicband_protocol.is_wand_packet(payload) - is_statue = (payload[0] == 0xC4 and len(payload) in (18, 23)) - if not (is_mb or is_wand or is_statue): - continue - now = time.monotonic() - if is_statue: - if now - state.last_statue_trigger < _STATUE_COOLDOWN_S: - continue - state.last_statue_trigger = now - else: - if (payload == state.last_payload - and now - state.last_payload_at < _DEDUP_WINDOW_S): - continue - state.last_payload = payload - state.last_payload_at = now - parsed = magicband_protocol.parse(payload) - candidate = renderer.for_command(parsed) - if candidate is not None: - new_command = (candidate, entry.rssi, payload) - finally: - adapter.stop_scan() - - # --- Helpers extracted in Stage D --- - _handle_button(state, button, batt, pixels) - - if new_command is not None: - state.active_state, rssi, raw = new_command - state.active_started_at = time.monotonic() - _log_command(state.active_state["label"], rssi, raw) - - if state.active_state is not None: - duration = state.active_state["duration_s"] - if duration is not None: - if time.monotonic() - state.active_started_at >= duration: - state.active_state = None - - _check_battery_log(state, batt) - _render_frame(zones, state, pixels) - pixels.show() - - elapsed = time.monotonic() - frame_start - remaining = _FRAME_BUDGET_S - elapsed - if remaining > 0: - time.sleep(remaining) - - -main() + elapsed = time.monotonic() - frame_start + remaining = _FRAME_BUDGET_S - elapsed + if remaining > 0: + time.sleep(remaining) diff --git a/BLE_Beacon_Ears/pixel_zones.py b/BLE_Beacon_Ears/pixel_zones.py index a75fe0c83..ecd88d0b0 100644 --- a/BLE_Beacon_Ears/pixel_zones.py +++ b/BLE_Beacon_Ears/pixel_zones.py @@ -3,11 +3,9 @@ # SPDX-License-Identifier: MIT '''Pixel zone abstraction for the BLE Beacon Ears project. -A "zone" represents one ear. In prototype mode (ONBOARD) both zones composite -into the single onboard NeoPixel, with the left zone painting first and the -right zone painting second so the right side "wins" - still useful for a -quick visual of current state. In production mode (STEREO_JEWELS) each zone -drives its own 7-pixel Jewel on an independent data pin. +A "zone" represents one ear. Each ear is driven by its own 7-pixel +Jewel on an independent data pin so the renderer can output stereo +effects (left-leads-right rotations, out-of-phase breathing). The API is intentionally minimal: zone.fill(rgb) - solid color across all pixels @@ -20,47 +18,9 @@ invisible to the eye. ''' # Target: Adafruit QT Py ESP32-S3 - the BLE Beacon Ears -import board -import digitalio import neopixel -class OnboardSingle: - '''Prototype mode: both zones share the single onboard NeoPixel. - - The left zone paints first; the right zone overwrites. This gives a - "right-biased" preview of stereo effects. Pair with the serial - stereo_log helper in the renderer to see what the other side is doing. - ''' - - def __init__(self, brightness=0.15): - self._power = None - try: - self._power = digitalio.DigitalInOut(board.NEOPIXEL_POWER) - self._power.switch_to_output(value=True) - except (AttributeError, ValueError): - pass - self._pixel = neopixel.NeoPixel( - board.NEOPIXEL, 1, brightness=brightness, auto_write=False) - self._pixel.fill((0, 0, 0)) - self._pixel.show() - - def make_zones(self): - '''Return (left_zone, right_zone) both pointing at the same pixel.''' - # Both zones share the underlying neopixel object. The renderer - # writes left first, then right. A single show() call at frame end - # flushes the right zone's final state to hardware. - return _SingleZone(self._pixel), _SingleZone(self._pixel) - - def show(self): - '''Flush the onboard pixel state to hardware.''' - self._pixel.show() - - def set_brightness(self, brightness): - '''Change pixel brightness at runtime.''' - self._pixel.brightness = brightness - - class StereoJewels: '''Production mode: two 7-pixel Jewels on independent data pins. @@ -120,29 +80,6 @@ def show(self): self._last_shown_black = False -class _SingleZone: - '''Zone backed by a shared 1-pixel NeoPixel (OnboardSingle mode).''' - - count = 1 - - def __init__(self, pixel): - self._pixel = pixel - - def fill(self, rgb): - '''Set the shared onboard pixel to the given color.''' - self._pixel[0] = rgb - - def set_led(self, idx, rgb): - '''Write a specific pixel index. Only idx 0 has effect here. - - All LED writes collapse to the single onboard pixel. The last - write of the frame wins, which matches the "right overwrites - left" compositing intent. - ''' - if idx == 0: - self._pixel[0] = rgb - - class _JewelZone: '''Zone backed by a dedicated 7-pixel Jewel (StereoJewels mode).'''