cloud/music_library
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[presto] elm-architecture-like POC

Browse Source
This commit is contained in:
Claudio Ortolina
2026-05-12 12:03:09 +01:00
parent dc5847bb43
commit dc64fb0bb0
1 changed files with 586 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
presto/poc.py
+586
View File
@@ -0,0 +1,586 @@
"""
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 = 210
ROW_1_Y = 110
ROW_2_Y = 170
ROW_3_Y = 230
ROW_4_Y = 290
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, 42)
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()