import os import time import board import wifi import socketpool import ssl import displayio import terminalio import adafruit_minimqtt.adafruit_minimqtt as MQTT import neopixel from adafruit_display_text import label from adafruit_matrixportal.matrixportal import MatrixPortal from adafruit_display_shapes.rect import Rect # --- SETUP --- displayio.release_displays() matrixportal = MatrixPortal(status_neopixel=board.NEOPIXEL) # Gauge config GAUGE_WIDTH = 60 GAUGE_HEIGHT = 6 GAUGE_X = 2 GAUGE_Y = 24 MAX_RPM = 11000 REDLINE_RPM = 9000 current_rpm = 0 target_rpm = 0 engine_state = "OFF" rpm_smoothing_rate = 1000 # adjust for animation speed #ref flash rev_flash_rect = Rect(0, 0, 64, 32, fill=0xFF0000) rev_flash_text = label.Label( terminalio.FONT, text="REV LIMIT!", scale=1, color=0xFFFFFF ) rev_flash_text.anchor_point = (0.5, 0.5) rev_flash_text.anchored_position = (32, 16) # Engine state rev_limiter_active = False last_flash_time = 0 last_rpm_displayed = 0 last_engine_state = None # UI Elements gauge_fill = Rect(GAUGE_X, GAUGE_Y, 1, GAUGE_HEIGHT, fill=0x00FF00) gauge_bg = Rect(GAUGE_X, GAUGE_Y, GAUGE_WIDTH, GAUGE_HEIGHT, fill=0x202020) rpm_label = label.Label(terminalio.FONT, text="", scale=1, color=0xFFFFFF) rpm_label.anchor_point = (0, 0) rpm_label.anchored_position = (0, 0) engine_off_label = label.Label(terminalio.FONT, text="ENGINE OFF", scale=1, color=0xFF0000) engine_off_label.anchor_point = (0.5, 0.5) engine_off_label.anchored_position = (32, 16) matrixportal.splash.append(gauge_bg) matrixportal.splash.append(gauge_fill) matrixportal.splash.append(rpm_label) # --- NETWORK --- aio_username = os.getenv("ADAFRUIT_AIO_USERNAME") aio_key = os.getenv("ADAFRUIT_AIO_KEY") broker = os.getenv("BROKER") port = int(os.getenv("PORT")) ssid = os.getenv("CIRCUITPY_WIFI_SSID") password = os.getenv("CIRCUITPY_WIFI_PASSWORD") print("Connecting to Wi-Fi...") wifi.radio.connect(ssid, password) print("Connected to", ssid) pool = socketpool.SocketPool(wifi.radio) mqtt_client = MQTT.MQTT( broker=broker, port=port, username=aio_username, password=aio_key, socket_pool=pool, ssl_context=ssl.create_default_context(), ) dashboard_feed = aio_username + "/feeds/text_box" def get_rpm_color(rpm): if rpm < 4000: return 0x00FF00 elif rpm < REDLINE_RPM: return 0xFFFF00 else: return 0xFF0000 # --- MQTT CALLBACKS --- def show_startup_animation(current_rpm): print(">> ENGINE STARTUP") # Green flash flash_rect = Rect(0, 0, 64, 32, fill=0x00FF00) flash_text = label.Label(terminalio.FONT, text="Cranking..", scale=1, color=0x000000) flash_text.anchor_point = (0.5, 0.5) flash_text.anchored_position = (32, 16) matrixportal.splash.append(flash_rect) matrixportal.splash.append(flash_text) time.sleep(1) matrixportal.splash.remove(flash_rect) matrixportal.splash.remove(flash_text) # RPM sweep animation for rpm in range(0, current_rpm + 1, int(current_rpm / 10) + 1): update_rpm_bar(rpm) time.sleep(0.05) def update_rpm_bar(rpm_value): global gauge_fill bar_width = max(1, int((rpm_value / MAX_RPM) * GAUGE_WIDTH)) fill_color = get_rpm_color(rpm_value) # Only update if bar size or color changed new_bar = Rect(GAUGE_X, GAUGE_Y, bar_width, GAUGE_HEIGHT, fill=fill_color) if gauge_fill in matrixportal.splash: matrixportal.splash.remove(gauge_fill) gauge_fill = new_bar matrixportal.splash.append(gauge_fill) def show_shutdown_animation(): print(">> ENGINE SHUTDOWN") flash_rect = Rect(0, 0, 64, 32, fill=0xFF0000) flash_text = label.Label(terminalio.FONT, text="ENGINE OFF", scale=1, color=0xFFFFFF) flash_text.anchor_point = (0.5, 0.5) flash_text.anchored_position = (32, 16) matrixportal.splash.append(flash_rect) matrixportal.splash.append(flash_text) time.sleep(1) matrixportal.splash.remove(flash_rect) matrixportal.splash.remove(flash_text) rpm_label.text = "" def connected(client, userdata, flags, rc): print("Connected to Adafruit IO. Subscribing...") client.subscribe(dashboard_feed) def message(client, topic, message): global gauge_fill, last_flash_time, rev_limiter_active, last_engine_state global target_rpm, engine_state,REDLINE_RPM print(f"Received: {message}") if " - " in message: try: rpm_part, engine_state_in = message.split(" - ") rpm = int(rpm_part.replace("RPM", "").strip()) engine_state_in = engine_state_in.strip() # Engine state change if engine_state_in != last_engine_state: if engine_state_in == "OFF": show_shutdown_animation() if engine_off_label not in matrixportal.splash: matrixportal.splash.append(engine_off_label) if gauge_fill in matrixportal.splash: matrixportal.splash.remove(gauge_fill) if rev_flash_rect in matrixportal.splash: matrixportal.splash.remove(rev_flash_rect) if rev_flash_text in matrixportal.splash: matrixportal.splash.remove(rev_flash_text) rpm_label.text = "" elif engine_state_in == "ON": show_startup_animation(rpm) if engine_off_label in matrixportal.splash: matrixportal.splash.remove(engine_off_label) last_engine_state = engine_state_in engine_state = engine_state_in # Update label and bar using current_rpm (which gets animated outside this function) if engine_state == "ON": rpm_label.text = f"RPM: {rpm}" now = time.monotonic() if rpm >= REDLINE_RPM: rev_limiter_active = True if now - last_flash_time > 0.15: last_flash_time = now if rev_limiter_active: rpm_bounced = max(0, rpm - 1000) bar_width = max(1, int((rpm_bounced / MAX_RPM) * GAUGE_WIDTH)) fill_color = 0xFFFFFF if rev_flash_rect not in matrixportal.splash: matrixportal.splash.append(rev_flash_rect) if rev_flash_text not in matrixportal.splash: matrixportal.splash.append(rev_flash_text) else: bar_width = max(1, int((rpm / MAX_RPM) * GAUGE_WIDTH)) fill_color = 0xFF0000 if rev_flash_rect in matrixportal.splash: matrixportal.splash.remove(rev_flash_rect) if rev_flash_text in matrixportal.splash: matrixportal.splash.remove(rev_flash_text) else: rev_limiter_active = False fill_color = get_rpm_color(rpm) bar_width = max(1, int((rpm / MAX_RPM) * GAUGE_WIDTH)) if rev_flash_rect in matrixportal.splash: matrixportal.splash.remove(rev_flash_rect) if rev_flash_text in matrixportal.splash: matrixportal.splash.remove(rev_flash_text) if gauge_fill in matrixportal.splash: matrixportal.splash.remove(gauge_fill) gauge_fill = Rect(GAUGE_X, GAUGE_Y, bar_width, GAUGE_HEIGHT, fill=fill_color) matrixportal.splash.append(gauge_fill) except Exception as e: print("Error parsing dashboard_text:", e) def disconnected(client, userdata, rc): print("Disconnected from Adafruit IO") mqtt_client.on_connect = connected mqtt_client.on_message = message mqtt_client.on_disconnect = disconnected print("Connecting to MQTT...") mqtt_client.connect() # --- LOOP --- while True: try: mqtt_client.loop() # Smoothly move toward target RPM if engine_state == "ON" and current_rpm != target_rpm: if abs(target_rpm - current_rpm) < rpm_smoothing_rate: current_rpm = target_rpm elif target_rpm > current_rpm: current_rpm += rpm_smoothing_rate else: current_rpm -= rpm_smoothing_rate update_rpm_bar(current_rpm) time.sleep(0.03) # time.sleep(1) for non-paid subscriptions except Exception as e: print("MQTT error:", e) time.sleep(5)