""" Elm-style architecture proof of concept for Pimoroni Presto. Shows WiFi status, an idle timer, and device time. Tapping the display flashes a small red circle at the touch point. """ import gc import time import network import ntptime from presto import Presto # ============================================================================ # HARDWARE # ============================================================================ presto = Presto() display = presto.display touch = presto.touch WIDTH, HEIGHT = display.get_bounds() # ============================================================================ # LAYOUT AND TIMING # ============================================================================ LOOP_SLEEP_MS = 30 TICK_MS = 1_000 WIFI_CHECK_MS = 5_000 WIFI_RETRY_MS = 30_000 WIFI_CONNECT_TIMEOUT_MS = 10_000 WIFI_CONNECT_POLL_MS = 250 FLASH_MS = 180 SCREEN_PADDING = 28 TITLE_Y = 28 LABEL_X = SCREEN_PADDING VALUE_X = 150 ROW_1_Y = 90 ROW_2_Y = 130 ROW_3_Y = 170 ROW_4_Y = 210 FOOTER_Y = HEIGHT - 50 FLASH_RADIUS = 14 # ============================================================================ # COLORS # ============================================================================ BG = (24, 24, 27) PANEL = (39, 39, 42) TITLE = (244, 244, 245) LABEL = (161, 161, 170) VALUE = (228, 228, 231) OK = (34, 197, 94) WARN = (250, 204, 21) ERROR = (239, 68, 68) DIM = (113, 113, 122) RED = (239, 68, 68) _pen_bg = None _pen_panel = None _pen_title = None _pen_label = None _pen_value = None _pen_ok = None _pen_warn = None _pen_error = None _pen_dim = None _pen_red = None # ============================================================================ # MESSAGES # ============================================================================ MSG_BOOT = 1 MSG_TICK = 2 MSG_TOUCH_DOWN = 3 MSG_TOUCH_UP = 4 MSG_FLASH_EXPIRED = 5 MSG_WIFI_CONNECTED = 6 MSG_WIFI_FAILED = 7 MSG_WIFI_STATUS = 8 MSG_TIME_SYNCED = 9 # ============================================================================ # EFFECTS # ============================================================================ FX_RENDER = 1 FX_CONNECT_WIFI = 2 FX_CHECK_WIFI = 3 FX_SYNC_TIME = 4 FX_COLLECT_GARBAGE = 5 # ============================================================================ # MODEL # ============================================================================ class Model: """Mutable application model. The POC keeps model updates cheap and explicit. Hardware interaction stays in effect functions, not in update(). """ def __init__(self): now = time.ticks_ms() self.now_ms = now self.started_ms = now self.last_activity_ms = now self.last_wifi_check_ms = 0 self.next_wifi_retry_ms = 0 self.wifi_status = "starting" self.wifi_detail = "" self.time_status = "not synced" self.current_time = format_device_time() self.idle_seconds = 0 self.flash_x = None self.flash_y = None self.flash_until_ms = 0 class Runtime: """Runtime-only event wiring state.""" def __init__(self): now = time.ticks_ms() self.next_tick_ms = now self.touch_down = False # ============================================================================ # UPDATE # ============================================================================ def update(model, msg): """Apply a message to the model and return effect commands.""" tag = msg[0] if tag == MSG_BOOT: model.wifi_status = "connecting" model.wifi_detail = "Looking for secrets.py" model.time_status = "waiting for WiFi" model.current_time = format_device_time() return [ (FX_RENDER,), (FX_CONNECT_WIFI,), ] if tag == MSG_TICK: now = msg[1] model.now_ms = now model.current_time = format_device_time() model.idle_seconds = time.ticks_diff(now, model.last_activity_ms) // 1000 commands = [(FX_RENDER,)] if should_retry_wifi(model, now): model.wifi_status = "connecting" model.wifi_detail = "Retrying" commands.append((FX_RENDER,)) commands.append((FX_CONNECT_WIFI,)) return commands if time.ticks_diff(now, model.last_wifi_check_ms) >= WIFI_CHECK_MS: model.last_wifi_check_ms = now commands.append((FX_CHECK_WIFI,)) return commands if tag == MSG_TOUCH_DOWN: x, y, now = msg[1], msg[2], msg[3] model.now_ms = now model.last_activity_ms = now model.idle_seconds = 0 model.flash_x = x model.flash_y = y model.flash_until_ms = time.ticks_add(now, FLASH_MS) return [(FX_RENDER,)] if tag == MSG_TOUCH_UP: return [] if tag == MSG_FLASH_EXPIRED: model.flash_x = None model.flash_y = None model.flash_until_ms = 0 return [(FX_RENDER,)] if tag == MSG_WIFI_CONNECTED: ip = msg[1] now = time.ticks_ms() model.wifi_status = "connected" model.wifi_detail = ip model.last_wifi_check_ms = now model.next_wifi_retry_ms = 0 model.time_status = "syncing" return [ (FX_RENDER,), (FX_SYNC_TIME,), (FX_COLLECT_GARBAGE,), ] if tag == MSG_WIFI_FAILED: reason = msg[1] now = time.ticks_ms() model.wifi_status = "failed" model.wifi_detail = reason model.last_wifi_check_ms = now model.next_wifi_retry_ms = time.ticks_add(now, WIFI_RETRY_MS) model.time_status = "not synced" return [ (FX_RENDER,), (FX_COLLECT_GARBAGE,), ] if tag == MSG_WIFI_STATUS: connected, ip = msg[1], msg[2] if connected: model.wifi_status = "connected" model.wifi_detail = ip model.next_wifi_retry_ms = 0 elif model.wifi_status == "connected": model.wifi_status = "disconnected" model.wifi_detail = "Will retry" model.next_wifi_retry_ms = time.ticks_add(time.ticks_ms(), WIFI_RETRY_MS) return [(FX_RENDER,)] if tag == MSG_TIME_SYNCED: ok = msg[1] model.current_time = format_device_time() model.time_status = "synced" if ok else "sync failed" return [(FX_RENDER,)] return [] def should_retry_wifi(model, now): if model.wifi_status == "connected" or model.wifi_status == "connecting": return False if model.next_wifi_retry_ms == 0: return False return time.ticks_diff(now, model.next_wifi_retry_ms) >= 0 # ============================================================================ # EVENT WIRING # ============================================================================ def wire_events(runtime, model): """Convert hardware polling and time into app messages.""" now = time.ticks_ms() messages = [] if time.ticks_diff(now, runtime.next_tick_ms) >= 0: runtime.next_tick_ms = time.ticks_add(now, TICK_MS) messages.append((MSG_TICK, now)) point = read_touch() if point is None: if runtime.touch_down: runtime.touch_down = False messages.append((MSG_TOUCH_UP, now)) elif not runtime.touch_down: runtime.touch_down = True messages.append((MSG_TOUCH_DOWN, point[0], point[1], now)) if model.flash_until_ms and time.ticks_diff(now, model.flash_until_ms) >= 0: messages.append((MSG_FLASH_EXPIRED,)) return messages # ============================================================================ # EFFECT RUNNER # ============================================================================ def dispatch(model, first_msg): """Process messages, run returned effects, and enqueue follow-up messages.""" messages = [first_msg] index = 0 while index < len(messages): msg = messages[index] index += 1 commands = update(model, msg) for command in commands: follow_up = run_effect(model, command) if follow_up: messages.extend(follow_up) def run_effect(model, command): kind = command[0] if kind == FX_RENDER: render(model) return [] if kind == FX_CONNECT_WIFI: return [connect_wifi_effect()] if kind == FX_CHECK_WIFI: return [check_wifi_effect()] if kind == FX_SYNC_TIME: return [sync_time_effect()] if kind == FX_COLLECT_GARBAGE: gc.collect() return [] return [] def connect_wifi_effect(): try: import secrets except ImportError: return (MSG_WIFI_FAILED, "secrets.py missing") ssid = getattr(secrets, "WIFI_SSID", "") password = getattr(secrets, "WIFI_PASSWORD", "") if not ssid: return (MSG_WIFI_FAILED, "WIFI_SSID missing") try: wlan = network.WLAN(network.STA_IF) wlan.active(True) if wlan.isconnected(): return (MSG_WIFI_CONNECTED, wlan.ifconfig()[0]) wlan.connect(ssid, password) start = time.ticks_ms() while not wlan.isconnected(): if time.ticks_diff(time.ticks_ms(), start) >= WIFI_CONNECT_TIMEOUT_MS: return (MSG_WIFI_FAILED, "connect timeout") sleep_ms(WIFI_CONNECT_POLL_MS) return (MSG_WIFI_CONNECTED, wlan.ifconfig()[0]) except Exception: return (MSG_WIFI_FAILED, "connect error") def check_wifi_effect(): try: wlan = network.WLAN(network.STA_IF) wlan.active(True) if wlan.isconnected(): return (MSG_WIFI_STATUS, True, wlan.ifconfig()[0]) return (MSG_WIFI_STATUS, False, "") except Exception: return (MSG_WIFI_STATUS, False, "") def sync_time_effect(): try: ntptime.settime() return (MSG_TIME_SYNCED, True) except Exception: return (MSG_TIME_SYNCED, False) # ============================================================================ # RENDER # ============================================================================ def render(model): """Imperative renderer. Drawing and display update are intentional effects.""" display.set_pen(_pen_bg) display.clear() draw_header() draw_status_rows(model) draw_flash(model) draw_footer() presto.update() def draw_header(): display.set_pen(_pen_title) display.set_font("bitmap14_outline") text = "Elm-style Presto POC" draw_centered_text(text, TITLE_Y, scale=1) def draw_status_rows(model): draw_row("WiFi", wifi_status_text(model), ROW_1_Y, wifi_pen(model)) draw_row("Detail", model.wifi_detail or "-", ROW_2_Y, _pen_value) draw_row("Idle", format_idle(model.idle_seconds), ROW_3_Y, _pen_value) draw_row("Time", model.current_time + " " + model.time_status, ROW_4_Y, _pen_value) def draw_row(label, value, y, value_pen): display.set_pen(_pen_panel) display.rectangle(SCREEN_PADDING, y - 12, WIDTH - 2 * SCREEN_PADDING, 22) display.set_font("bitmap8") display.set_pen(_pen_label) display.text(label, LABEL_X, y, scale=1) display.set_pen(value_pen) display.text(value, VALUE_X, y, scale=1) def draw_flash(model): if model.flash_x is None or model.flash_y is None: return display.set_pen(_pen_red) try: display.circle(model.flash_x, model.flash_y, FLASH_RADIUS) except Exception: display.rectangle( model.flash_x - FLASH_RADIUS, model.flash_y - FLASH_RADIUS, FLASH_RADIUS * 2, FLASH_RADIUS * 2 ) def draw_footer(): display.set_pen(_pen_dim) display.set_font("bitmap8") draw_centered_text("Tap anywhere", FOOTER_Y, scale=1) def draw_centered_text(text, y, scale=1): width = display.measure_text(text, scale=scale) display.text(text, (WIDTH - width) // 2, y, scale=scale) def wifi_status_text(model): if model.wifi_status == "connected": return "Connected" if model.wifi_status == "connecting": return "Connecting" if model.wifi_status == "disconnected": return "Disconnected" if model.wifi_status == "failed": return "Failed" return "Starting" def wifi_pen(model): if model.wifi_status == "connected": return _pen_ok if model.wifi_status == "connecting": return _pen_warn return _pen_error # ============================================================================ # HARDWARE HELPERS # ============================================================================ def init_display(): global _pen_bg, _pen_panel, _pen_title, _pen_label, _pen_value global _pen_ok, _pen_warn, _pen_error, _pen_dim, _pen_red _pen_bg = display.create_pen(*BG) _pen_panel = display.create_pen(*PANEL) _pen_title = display.create_pen(*TITLE) _pen_label = display.create_pen(*LABEL) _pen_value = display.create_pen(*VALUE) _pen_ok = display.create_pen(*OK) _pen_warn = display.create_pen(*WARN) _pen_error = display.create_pen(*ERROR) _pen_dim = display.create_pen(*DIM) _pen_red = display.create_pen(*RED) try: display.set_font("bitmap8") except Exception: pass try: presto.set_backlight(1.0) except Exception: pass def read_touch(): try: touch.poll() except Exception: pass try: if touch.state: return int(touch.x), int(touch.y) except Exception: pass for method_name in ("read", "get_touch"): method = getattr(touch, method_name, None) if method is None: continue try: point = method() except Exception: continue normalised = normalise_touch(point) if normalised is not None: return normalised return None def normalise_touch(point): if not point: return None if isinstance(point, (tuple, list)): if len(point) == 2: return int(point[0]), int(point[1]) if len(point) >= 3 and point[0] is not None: return int(point[1]), int(point[2]) return None try: if hasattr(point, "state") and not point.state: return None return int(point.x), int(point.y) except Exception: return None # ============================================================================ # FORMAT HELPERS # ============================================================================ def format_device_time(): lt = time.localtime() return "{:02d}:{:02d}:{:02d}".format(lt[3], lt[4], lt[5]) def format_idle(seconds): minutes = seconds // 60 remaining = seconds % 60 return "{:02d}:{:02d}".format(minutes, remaining) def sleep_ms(ms): try: time.sleep_ms(ms) except AttributeError: time.sleep(ms / 1000.0) # ============================================================================ # MAIN # ============================================================================ def main(): init_display() model = Model() runtime = Runtime() dispatch(model, (MSG_BOOT,)) while True: messages = wire_events(runtime, model) for msg in messages: dispatch(model, msg) sleep_ms(LOOP_SLEEP_MS) main()