ecd1d2a189
- Built-in Song playlist: every section is now b4 (~4 bars) so Continue rolls one into the next quickly. - On-screen timer now counts WITHIN the current segment and resets every time the bar counter wraps (new _seg_start, reset at each b<n> boundary + on _reset_clock). The practice-log duration still uses play_start (total). Unified the segment-boundary handling (timer reset + ramp restart + Continue advance) in _on_new_bar. - MIDI stutter: display.refresh() BLOCKS on the SPI stream and was delaying the next beat's note. Cap refresh to ~30Hz and poll the GT911 touch ~30Hz (was every loop) so the scheduler fires notes on time; visuals lag a few ms (imperceptible). Verified in harness: Build(b4,rmp92/4/2) bpm 92->96->reset@bar4, seg_start resets only at the boundary, Continue arms there; edit tests pass; app.mpy builds (C/v6). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1014 lines
56 KiB
Python
1014 lines
56 KiB
Python
# VARASYS PolyMeter - PM_K-1 "Kit" firmware (CircuitPython edition)
|
|
# Raspberry Pi Pico (Pico / Pico W / Pico 2) on the 52Pi EP-0172 "Pico Breadboard Kit Plus":
|
|
# 3.5" ST7796 320x480 cap-touch (GT911), PSP joystick, WS2812 RGB, buzzer, 2 buttons.
|
|
#
|
|
# WHY CIRCUITPYTHON: the board then mounts as a USB drive (CIRCUITPY) carrying this code, your
|
|
# tracks (programs.json) and a copy of the editor - edit on the web, "Save to device" writes
|
|
# programs.json here, and CircuitPython auto-reloads with the new grooves. It also sends USB-MIDI
|
|
# (a note per click) so the web editor can play it out the computer's speakers ("Device audio").
|
|
# Runs the SAME program strings as metronome.varasys.io.
|
|
#
|
|
# INSTALL: flash CircuitPython (https://circuitpython.org/board/raspberry_pi_pico/), then copy
|
|
# this file as code.py plus programs.json onto the CIRCUITPY drive. It runs on boot.
|
|
#
|
|
# Fallback: the simpler MicroPython firmware (pico/main.py) is always available - BOOTSEL +
|
|
# drag a MicroPython .uf2 to go back. The Pico cannot be bricked.
|
|
#
|
|
# Untested-panel notes & calibration flags are in CONFIG + pico-cp/README.md.
|
|
|
|
import board, busio, digitalio, analogio, pwmio, displayio, vectorio, time, json, gc, os, supervisor
|
|
supervisor.runtime.autoreload = False # we write our own files (log + pushed programs); never self-restart
|
|
APP_VERSION = "0.0.12" # firmware version (the A/B updater pushes/compares this)
|
|
try:
|
|
import rtc # set from the editor's clock SysEx so the log has real timestamps
|
|
except ImportError:
|
|
rtc = None
|
|
try: # CircuitPython 9.x
|
|
from fourwire import FourWire
|
|
from busdisplay import BusDisplay
|
|
except ImportError: # CircuitPython 8.x
|
|
from displayio import FourWire
|
|
from displayio import Display as BusDisplay
|
|
try:
|
|
import neopixel_write # core module on RP2040 - drives WS2812 with no external library
|
|
except ImportError:
|
|
neopixel_write = None
|
|
try:
|
|
import usb_midi # default-enabled on RP2040 - sends a MIDI note per click to the computer
|
|
except ImportError:
|
|
usb_midi = None
|
|
try:
|
|
from binascii import a2b_base64 # decode the base64-encoded .mpy pushed by the editor's one-click update
|
|
except ImportError:
|
|
a2b_base64 = None
|
|
|
|
# ============================== CONFIG (tweak if needed) ==============================
|
|
SPI_BAUD = 62_500_000 # faster SPI = smaller tearing window; drop to 40_000_000 if unstable
|
|
LED_BRIGHTNESS = 0.15 # WS2812 sits right next to you - keep it dim (0..1)
|
|
MIDI_ENABLED = True # send a USB-MIDI note per click (play via the web editor's "Device audio")
|
|
MUTE_BUZZER = False # silence the on-board buzzer (e.g. when using computer audio)
|
|
WIDTH, HEIGHT = 320, 480
|
|
MADCTL = 0x48 # portrait; 0x48 swaps R/B for this BGR panel (cyan reads cyan). Use 0x40 if reversed.
|
|
INVERT_COLORS = True # most ST7796 modules need inversion ON; set False if colours look negative
|
|
# Touch (GT911) - flip if taps land wrong:
|
|
TOUCH_SWAP_XY = False
|
|
TOUCH_INVERT_X = False
|
|
TOUCH_INVERT_Y = False
|
|
TOUCH_DEBUG = False
|
|
# Joystick:
|
|
JOY_INVERT_X = False
|
|
JOY_INVERT_Y = False
|
|
JOY_DEADZONE = 9000
|
|
|
|
# ----- pins (fixed by the EP-0172 board) -----
|
|
P_SCK, P_MOSI, P_CS, P_DC, P_RST = board.GP2, board.GP3, board.GP5, board.GP6, board.GP7
|
|
P_SDA, P_SCL = board.GP8, board.GP9
|
|
P_RGB, P_BUZ, P_BTNA, P_BTNB = board.GP12, board.GP13, board.GP15, board.GP14
|
|
P_JOYX, P_JOYY = board.GP26, board.GP27
|
|
|
|
# ----- BUILT-IN playlists: the standard defaults from the web editor, baked in here so they update with
|
|
# firmware and the user can't change/delete them. User playlists live separately in programs.json
|
|
# (pushed from the editor) and never touch these. (ASCII only - it's pushed 7-bit + the fonts are ASCII.)
|
|
BUILTIN_SETLISTS = [
|
|
("Styles", [
|
|
("Four-on-the-floor", "t120;kick:4;snare:4=.x.x;hatClosed:4/2"),
|
|
("Swing ride", "t150;ride:4/2s;kick:4=X..x;snare:4=.x.x"),
|
|
("Purdie half-time shuffle", "t92;kick:4/3=X....x...x..;snare:4/3=..gg.gX.gg.g;hatClosed:4/3=X.xX.xX.xX.x"),
|
|
("Samba (2/4)", "t104;tomLow:2/4=x...X...;hatClosed:2/4;woodblock:2/4=X.xx.xX."),
|
|
("Nanigo (6/8 bembe)", "t130;cowbell:4/3=X.xx.x.xx.x.;kick:4/3=X.....X.....;hatClosed:4/3=..x..x..x..x"),
|
|
("6/8 groove", "t100;kick:3+3=x..x..;snare:3+3=...x..;hatClosed:3+3/2"),
|
|
("7/8 (2+2+3)", "t130;kick:2+2+3=x..x..x;hatClosed:2+2+3/2"),
|
|
("5/4 (3+2)", "t112;kick:3+2=x..x.;snare:3+2=..x..;hatClosed:3+2/2"),
|
|
]),
|
|
("Practice", [
|
|
("5 over 4 polyrhythm", "t100;kick:4;claves:5~"),
|
|
("3 over 2 hemiola", "t96;woodblock:2;cowbell:3~"),
|
|
("2 & 4 & 3 over one bar", "t100;kick:3;cowbell:2~;claves:4~"),
|
|
("Triplet hats", "t100;kick:4;snare:4=.x.x;hatClosed:4/3"),
|
|
("Accents - cycle the pads", "t92;kick:4=X..X;snare:4=.X.X;hatClosed:4/2"),
|
|
("Tempo builder 80 up", "t80;woodblock:4;rmp80/4/4"),
|
|
("Gap trainer (play 2 / rest 2)", "t100;kick:4;hatClosed:4/2;tr2/2"),
|
|
]),
|
|
("Song (continuous)", [ # ~4-bar sections; with Continue on they roll one into the next
|
|
("Intro - hats & kick", "t88;b4;kick:4=X.x.;hatClosed:4/2=gggggggg"),
|
|
("Groove in - backbeat", "t88;b4;kick:4=X.x.;snare:4=.X.X;hatClosed:4/2"),
|
|
("Half-time shuffle", "t92;b4;kick:4/3=X....x...x..;snare:4/3=..gg.gX.gg.g;hatClosed:4/3=X.xX.xX.xX.x"),
|
|
("Build - ramp 92-120", "t92;b4;rmp92/4/2;kick:4;snare:4=.X.X;hatClosed:4/2"),
|
|
("Four-on-the-floor (909)", "t124;b4;kick909:4;clap909:4=.X.X;hat909:4/2=.X.X.X.X"),
|
|
("Samba break (2/4)", "t116;b4;tomLow:2/4=x...X...;hatClosed:2/4;woodblock:2/4=X.xx.xX."),
|
|
("Peak - 16ths", "t132;b4;kick:4=X..x;snare:4=.X.X;hatClosed:4/4"),
|
|
("Outro - ramp down", "t132;b4;rmp132/-7/1;kick:4=X..x;hatClosed:4/2=gggggggg"),
|
|
]),
|
|
]
|
|
|
|
# ============================== COLOURS (0xRRGGBB; displayio handles 565) ==============================
|
|
C_BG, C_PANEL, C_TXT, C_MUTE = 0x06090E, 0x1C222C, 0xC7D0DB, 0x788494
|
|
C_CYAN, C_AMBER, C_GREEN, C_DIM = 0x0AB3F7, 0xFF9B2E, 0x2FE07A, 0x243240
|
|
C_BTN = 0x1C222C
|
|
LEVEL_RGB = {2: (255, 110, 0), 1: (0, 150, 255), 3: (130, 70, 255)} # beat pulse: accent / normal / ghost
|
|
LED_IDLE = (0, 80, 0) # RGB LED resting colour when stopped: dim green ("on")
|
|
LED_RUN = (110, 0, 0) # RGB LED resting colour when playing: dim red (beats pulse brighter on top)
|
|
# voice -> General-MIDI note (USB-MIDI bridge), and level -> MIDI velocity
|
|
SOUND_GM = {"kick":36,"kick808":36,"kick909":36, "snare":38,"snare808":38,"snare909":38,
|
|
"clap":39,"clap808":39,"clap909":39, "rim":37, "hatClosed":42,"hat808":42,"hat909":42,
|
|
"hatOpen":46,"openHat808":46, "ride":51,"ride909":51, "crash":49,"crash909":49,
|
|
"tomLow":41,"tom808":45,"tomMid":45,"tomHigh":48, "tambourine":54,
|
|
"cowbell":56,"cowbell808":56, "woodblock":76,"jamblock":76, "claves":75, "beep":37}
|
|
GM_DEFAULT = 37
|
|
MIDI_VEL = {2: 120, 1: 90, 3: 45} # accent / normal / ghost
|
|
MAXLANES = 5 # lanes shown on the pad grid (extras still play)
|
|
GRID_TOP = 158 # top of the pad grid (leaves room for time/bar/ramp/tab rows)
|
|
LOG_TOP, LOG_ROWH, LOG_ROWS = 302, 16, 9 # practice-history log area (below the pad grid)
|
|
MIN_LOG_SEC = 5 # don't log plays shorter than this
|
|
PAD_DIM = (0x10161E, 0x0A3A52, 0x4A3010, 0x2A1D4A) # idle pad: mute / normal / accent / ghost
|
|
PAD_LIT = (0x39414D, 0x0AB3F7, 0xFF9B2E, 0x967BFF) # playhead pad: mute / normal / accent / ghost
|
|
C_GRID = 0x1A2330 # faint vertical beat gridlines (beats line up across lanes)
|
|
C_RED = 0xFF5A5A # unsaved-edits (dirty) track title
|
|
|
|
# WS2812 RGB LED - self-contained via the core neopixel_write module (no external library)
|
|
class RGB:
|
|
def __init__(self, pin):
|
|
self.ok = neopixel_write is not None
|
|
if self.ok:
|
|
self.io = digitalio.DigitalInOut(pin); self.io.direction = digitalio.Direction.OUTPUT
|
|
self.buf = bytearray(3)
|
|
def set(self, r, g, b):
|
|
if not self.ok: return
|
|
# WS2812 wants GRB order; scale down so it isn't blinding
|
|
self.buf[0] = int(g * LED_BRIGHTNESS); self.buf[1] = int(r * LED_BRIGHTNESS); self.buf[2] = int(b * LED_BRIGHTNESS)
|
|
try: neopixel_write.neopixel_write(self.io, self.buf)
|
|
except Exception: self.ok = False
|
|
|
|
# ============================== ANTI-ALIASED FONTS (binary blobs on the drive; see pico/gen_font.py) ==============================
|
|
def load_font(path):
|
|
with open(path, "rb") as f:
|
|
blob = f.read()
|
|
count = blob[0]; p = 1; pixoff = 1 + count * 7; glyphs = {}
|
|
for _ in range(count):
|
|
cp = (blob[p] << 8) | blob[p+1]; w = blob[p+2]; h = blob[p+3]
|
|
xoff = blob[p+4]; xoff = xoff - 256 if xoff > 127 else xoff
|
|
top = blob[p+5]; adv = blob[p+6]; p += 7
|
|
glyphs[cp] = (w, h, xoff, top, adv, pixoff); pixoff += (w * h + 1) // 2
|
|
return (glyphs, blob)
|
|
|
|
FONT_S = load_font("/font_s.bin") # small - pad-grid lane labels
|
|
FONT_M = load_font("/font_m.bin") # labels / buttons
|
|
FONT_L = load_font("/font_l.bin") # big BPM
|
|
gc.collect()
|
|
|
|
def _blend(bg, fg, i):
|
|
t = i * 17
|
|
r = (((bg >> 16) & 0xFF)*(255-t) + ((fg >> 16) & 0xFF)*t) // 255
|
|
g = (((bg >> 8) & 0xFF)*(255-t) + ((fg >> 8) & 0xFF)*t) // 255
|
|
b = ((bg & 0xFF)*(255-t) + (fg & 0xFF)*t) // 255
|
|
return (r << 16) | (g << 8) | b
|
|
|
|
def make_text(s, font, fg, bg):
|
|
"""Render a string into a displayio TileGrid (anti-aliased via a 16-step blend palette)."""
|
|
glyphs, blob = font
|
|
w = 0; top0 = 999; bot = 0
|
|
for c in s:
|
|
g = glyphs.get(ord(c))
|
|
if not g: continue
|
|
w += g[4]
|
|
if g[1]:
|
|
if g[3] < top0: top0 = g[3]
|
|
if g[3] + g[1] > bot: bot = g[3] + g[1]
|
|
if top0 == 999: top0 = 0
|
|
w = max(1, w); h = max(1, bot - top0)
|
|
gc.collect()
|
|
bmp = displayio.Bitmap(w, h, 16)
|
|
pal = displayio.Palette(16)
|
|
for i in range(16): pal[i] = _blend(bg, fg, i)
|
|
pen = 0
|
|
for c in s:
|
|
g = glyphs.get(ord(c))
|
|
if not g: continue
|
|
gw, gh, xoff, gtop, adv, off = g
|
|
for j in range(gh):
|
|
row = (gtop - top0) + j
|
|
for i in range(gw):
|
|
k = j * gw + i
|
|
byte = blob[off + (k >> 1)]
|
|
nib = (byte >> 4) if (k & 1) == 0 else (byte & 0xF)
|
|
if nib:
|
|
x = pen + xoff + i
|
|
if 0 <= x < w and 0 <= row < h: bmp[x, row] = nib
|
|
pen += adv
|
|
return displayio.TileGrid(bmp, pixel_shader=pal), w, h
|
|
|
|
# ---- single-image alpha assets (logo, status icons) - blit like a one-off glyph; see gen_assets.py ----
|
|
def load_alpha(path):
|
|
try:
|
|
with open(path, "rb") as f: blob = f.read()
|
|
return (blob[0], blob[1], blob) # (w, h, bytes); pixels start at offset 2
|
|
except Exception:
|
|
return None # missing/corrupt -> caller falls back to text (no crash)
|
|
def make_glyph(asset, fg, bg):
|
|
w, h, blob = asset
|
|
gc.collect()
|
|
bmp = displayio.Bitmap(w, h, 16); pal = displayio.Palette(16)
|
|
for i in range(16): pal[i] = _blend(bg, fg, i)
|
|
for k in range(w * h):
|
|
byte = blob[2 + (k >> 1)]
|
|
nib = (byte >> 4) if (k & 1) == 0 else (byte & 0xF)
|
|
if nib: bmp[k % w, k // w] = nib
|
|
return displayio.TileGrid(bmp, pixel_shader=pal), pal, w, h
|
|
def _recolor(pal, fg, bg): # re-tint a stored asset palette in place (tear-free)
|
|
for i in range(16): pal[i] = _blend(bg, fg, i)
|
|
|
|
LOGO = load_alpha("/logo.bin") # VARASYS wordmark (no tagline)
|
|
ICON_MIDI = load_alpha("/midi.bin") # DIN-5: green when a MIDI host is listening
|
|
ICON_USB = load_alpha("/usb.bin") # trident: lit when USB-connected to a computer
|
|
gc.collect()
|
|
|
|
# ============================== POLYMETER ENGINE (same semantics as the web/MicroPython) ==============================
|
|
PAT = {'X': 2, 'x': 1, 'g': 3, '.': 0, '-': 0, '_': 0}
|
|
PRIO = {2: 3, 1: 2, 3: 1}
|
|
|
|
def parse_program(s):
|
|
bpm = 120; lanes = []; bars = 0; ramp = None; trainer = None
|
|
for tok in s.strip().split(';'):
|
|
tok = tok.strip()
|
|
if not tok: continue
|
|
if tok[0] == 't' and tok[1:].isdigit():
|
|
bpm = int(tok[1:]); continue
|
|
if tok[0] == 'b' and tok[1:].isdigit(): # b<n> = segment length in bars (totals + Continue)
|
|
bars = int(tok[1:]); continue # (lane sounds like "beep:4" have a ':' -> not matched here)
|
|
if tok.startswith('rmp'): # rmp<start>/<amount>/<everyBars> tempo ramp (amount may be -)
|
|
p = tok[3:].split('/')
|
|
if len(p) == 3:
|
|
try: ramp = {'start': int(p[0]), 'amt': int(p[1]), 'every': max(1, int(p[2]))}
|
|
except ValueError: pass
|
|
continue
|
|
if tok.startswith('tr') and '/' in tok and ':' not in tok: # tr<play>/<mute> gap trainer (bars)
|
|
p = tok[2:].split('/')
|
|
if len(p) == 2:
|
|
try: trainer = {'play': max(0, int(p[0])), 'mute': max(0, int(p[1]))}
|
|
except ValueError: pass
|
|
continue
|
|
if ':' not in tok: continue
|
|
lane = _parse_lane(tok)
|
|
if lane: lanes.append(lane)
|
|
if not lanes: lanes = [_parse_lane("beep:4")]
|
|
return max(30, min(300, bpm)), lanes, bars, ramp, trainer
|
|
|
|
def _parse_lane(tok):
|
|
poly = '~' in tok; mute = '!' in tok
|
|
tok = tok.replace('~', '').replace('!', '')
|
|
gain = ''
|
|
if '@' in tok: tok, _, g = tok.partition('@'); gain = '@' + g # preserve @db for round-trip (engine ignores it)
|
|
sound, _, rest = tok.partition(':')
|
|
pattern = None
|
|
if '=' in rest: rest, _, pattern = rest.partition('=')
|
|
sub = 1; swing = False
|
|
if '/' in rest:
|
|
rest, _, sd = rest.partition('/')
|
|
swing = sd.endswith('s'); sd = sd.rstrip('s') # "/2s" = swung eighths
|
|
sub = int(sd) if sd.isdigit() else 1
|
|
groups = [int(g) for g in rest.split('+') if g.isdigit()] or [4]
|
|
beats = sum(groups); starts = set(); acc = 0
|
|
for gp in groups: starts.add(acc); acc += gp
|
|
steps = beats * sub
|
|
if pattern:
|
|
levels = [PAT.get(ch, 0) for ch in pattern]
|
|
if len(levels) < steps: levels += [0] * (steps - len(levels))
|
|
steps = len(levels)
|
|
else:
|
|
levels = []
|
|
for i in range(steps):
|
|
if i % sub == 0: levels.append(2 if (i // sub) in starts else 1)
|
|
else: levels.append(0)
|
|
return {'sound': sound, 'sub': sub, 'swing': swing, 'steps': steps, 'levels': levels,
|
|
'poly': poly, 'mute': mute, 'groups': groups, 'gain': gain}
|
|
|
|
PAT_CH = {2: 'X', 1: 'x', 3: 'g', 0: '.'} # level -> pattern char (inverse of PAT)
|
|
def lane_to_str(L): # serialize a lane back to the share grammar (round-trips)
|
|
s = L['sound'] + ':' + '+'.join(str(g) for g in L.get('groups', [4]))
|
|
if L['sub'] != 1 or L['swing']: s += '/' + str(L['sub']) + ('s' if L['swing'] else '')
|
|
s += '=' + ''.join(PAT_CH.get(v, '.') for v in L['levels'])
|
|
s += L.get('gain', '')
|
|
if L['poly']: s += '~'
|
|
if L['mute']: s += '!'
|
|
return s
|
|
|
|
_ALNUM = "abcdefghijklmnopqrstuvwxyz0123456789"
|
|
def _slkey(t): # normalise a title for built-in/user de-dup (no str.isalnum on CircuitPython)
|
|
return "".join(c for c in t.lower() if c in _ALNUM)
|
|
def load_user_setlists():
|
|
# User playlists from /programs.json (pushed by the editor). New {setlists:[{title,programs:[..]}]} form,
|
|
# or the old flat {programs:[..]} (one list). Built-ins are baked in BUILTIN_SETLISTS, never here.
|
|
try:
|
|
with open("/programs.json") as f: d = json.load(f)
|
|
except Exception as e:
|
|
print("programs.json:", e); return []
|
|
def items_of(pl): return [(p.get("name", "?"), p.get("prog", "")) for p in pl if p.get("prog")]
|
|
out = []
|
|
try:
|
|
if isinstance(d.get("setlists"), list):
|
|
for sl in d["setlists"]:
|
|
it = items_of(sl.get("programs", []))
|
|
if it: out.append((sl.get("title", "My set list"), it))
|
|
elif isinstance(d.get("programs"), list):
|
|
it = items_of(d["programs"])
|
|
if it: out.append((d.get("title", "My set list"), it))
|
|
except Exception as e:
|
|
print("setlists:", e)
|
|
return out
|
|
|
|
# ============================== GT911 TOUCH ==============================
|
|
class GT911:
|
|
def __init__(self, i2c):
|
|
self.i2c = i2c; self.addr = None
|
|
while not i2c.try_lock(): pass
|
|
try: found = i2c.scan()
|
|
finally: i2c.unlock()
|
|
for a in (0x5D, 0x14):
|
|
if a in found: self.addr = a; break
|
|
if self.addr is None and found: self.addr = found[0]
|
|
def _rd(self, reg, n):
|
|
b = bytearray(n)
|
|
while not self.i2c.try_lock(): pass
|
|
try:
|
|
self.i2c.writeto(self.addr, bytes([reg >> 8, reg & 0xFF]))
|
|
self.i2c.readfrom_into(self.addr, b)
|
|
finally: self.i2c.unlock()
|
|
return b
|
|
def _wr(self, reg, val):
|
|
while not self.i2c.try_lock(): pass
|
|
try: self.i2c.writeto(self.addr, bytes([reg >> 8, reg & 0xFF, val]))
|
|
finally: self.i2c.unlock()
|
|
def read(self):
|
|
if self.addr is None: return None
|
|
try: st = self._rd(0x814E, 1)[0]
|
|
except OSError: return None
|
|
if not (st & 0x80): return None
|
|
n = st & 0x0F; pt = None
|
|
if n >= 1:
|
|
b = self._rd(0x8150, 4); tx = b[0] | (b[1] << 8); ty = b[2] | (b[3] << 8)
|
|
pt = self._map(tx, ty)
|
|
try: self._wr(0x814E, 0)
|
|
except OSError: pass
|
|
return pt
|
|
def _map(self, tx, ty):
|
|
if TOUCH_DEBUG: print("touch raw", tx, ty)
|
|
if TOUCH_SWAP_XY: tx, ty = ty, tx
|
|
if TOUCH_INVERT_X: tx = WIDTH - 1 - tx
|
|
if TOUCH_INVERT_Y: ty = HEIGHT - 1 - ty
|
|
if 0 <= tx < WIDTH and 0 <= ty < HEIGHT: return (tx, ty)
|
|
return None
|
|
|
|
# ============================== DISPLAY SETUP ==============================
|
|
def st7796_init():
|
|
inv = b'\x21\x00' if INVERT_COLORS else b'\x20\x00'
|
|
return (
|
|
b'\x01\x80\x78' # SWRESET + 120ms
|
|
b'\x11\x80\x78' # SLPOUT + 120ms
|
|
b'\xF0\x01\xC3' b'\xF0\x01\x96' # command-set unlock
|
|
+ bytes([0x36, 0x01, MADCTL]) +
|
|
b'\x3A\x01\x55' # 16bpp
|
|
b'\xB4\x01\x01'
|
|
b'\xB6\x03\x80\x02\x3B'
|
|
b'\xE8\x08\x40\x8A\x00\x00\x29\x19\xA5\x33'
|
|
b'\xC1\x01\x06' b'\xC2\x01\xA7'
|
|
b'\xC5\x81\x18\x78' # VCOM + 120ms
|
|
b'\xE0\x0E\xF0\x09\x0B\x06\x04\x15\x2F\x54\x42\x3C\x17\x14\x18\x1B'
|
|
b'\xE1\x0E\xE0\x09\x0B\x06\x04\x03\x2B\x43\x42\x3B\x16\x14\x17\x1B'
|
|
b'\xF0\x01\x3C' b'\xF0\x81\x69\x78' # lock + 120ms
|
|
+ inv +
|
|
b'\x29\x80\x32' # DISPON + 50ms
|
|
)
|
|
|
|
def make_display():
|
|
displayio.release_displays()
|
|
spi = busio.SPI(clock=P_SCK, MOSI=P_MOSI)
|
|
bus = FourWire(spi, command=P_DC, chip_select=P_CS, reset=P_RST, baudrate=SPI_BAUD)
|
|
return BusDisplay(bus, st7796_init(), width=WIDTH, height=HEIGHT, auto_refresh=False)
|
|
|
|
def solid(color):
|
|
p = displayio.Palette(1); p[0] = color; return p
|
|
|
|
def rect(x, y, w, h, color):
|
|
return vectorio.Rectangle(pixel_shader=solid(color), width=w, height=h, x=x, y=y)
|
|
|
|
# ============================== APP ==============================
|
|
class App:
|
|
def __init__(self):
|
|
self.display = make_display()
|
|
self.i2c = busio.I2C(scl=P_SCL, sda=P_SDA, frequency=400_000)
|
|
self.touch = GT911(self.i2c)
|
|
self.midi = usb_midi.ports[1] if (MIDI_ENABLED and usb_midi and len(usb_midi.ports) > 1) else None
|
|
self.midi_in = usb_midi.ports[0] if (MIDI_ENABLED and usb_midi and len(usb_midi.ports) > 0) else None
|
|
self._mbuf = bytearray(64); self.midi_host = False; self.last_midi_in = 0.0
|
|
self._sx = bytearray(); self._sxon = False # USB-MIDI SysEx assembler (clock + pushed programs)
|
|
self._fw = None # chunked firmware transfer: staging file handle
|
|
self.led = RGB(P_RGB)
|
|
self.buz = pwmio.PWMOut(P_BUZ, frequency=1600, variable_frequency=True, duty_cycle=0)
|
|
self.buz_off = 0
|
|
self.btnA = self._btn(P_BTNA); self.btnB = self._btn(P_BTNB)
|
|
self._aPrev = True; self._bPrev = True
|
|
self.jx = analogio.AnalogIn(P_JOYX); self.jy = analogio.AnalogIn(P_JOYY)
|
|
self._joyNext = 0
|
|
self._touchDown = False; self._touchSeen = 0
|
|
self.running = False; self.bpm = 120; self.idx = 0; self.lanes = []; self.bars = 0; self.rgb = (0, 0, 0)
|
|
self.ramp = None; self.trainer = None; self._lastbar = -1; self._muted = False; self._ramp_base = 120
|
|
self._dirty = False; self._overlay = None; self._ovbtns = [] # on-device editing: unsaved edits + modal
|
|
self.continue_on = False; self._advance = False; self._grid = {} # auto-advance + pad hit-test geometry
|
|
self.sl = 0; self.rebuild_setlists() # built-in playlists (baked) + user playlists (programs.json)
|
|
self.dirty = True
|
|
self.pad_pal = displayio.Palette(8) # 0-3 idle levels (mute/normal/accent/ghost), 4-7 the lit playhead
|
|
for i in range(4): self.pad_pal[i] = PAD_DIM[i]; self.pad_pal[i + 4] = PAD_LIT[i]
|
|
self.lane_pads = []; self.lane_lit = []
|
|
self.usb_conn = False; self._m_steps = 0 # USB-connected state; master-lane steps (for the bar counter)
|
|
self._uiNext = 0.0; self._lastTs = None; self._lastBs = None # throttle the stopwatch/bar redraw
|
|
self._seg_start = 0.0 # timer origin; resets with the bar counter (each segment)
|
|
self._refreshNext = 0.0; self._touchNext = 0.0 # cap display refresh + touch polling (tighter MIDI timing)
|
|
self.ic_midi_pal = None; self.ic_usb_pal = None
|
|
# practice history - persisted to /history.json (next to programs.json) when we own the filesystem
|
|
self.can_write = self._probe_write()
|
|
self.log = self._load_log()
|
|
self.play_start = None; self.play_bpm = 0; self.play_name = ""
|
|
self._armed = None; self.log_rows = []
|
|
self._build_scene()
|
|
self.load(0) # load() also draws the (track-filtered) practice log
|
|
self.draw_icons(); self.draw_meters(); self.led_rest() # LED green = on
|
|
|
|
def _btn(self, pin):
|
|
d = digitalio.DigitalInOut(pin); d.direction = digitalio.Direction.INPUT; d.pull = digitalio.Pull.UP
|
|
return d
|
|
|
|
# ---------- scene graph ----------
|
|
def _build_scene(self):
|
|
root = displayio.Group(); self.display.root_group = root
|
|
root.append(rect(0, 0, WIDTH, HEIGHT, C_BG)) # static background (run state shows on the LED)
|
|
# header: VARASYS logo (left, no tagline) + version (small, top, right of the logo) + MIDI/USB icons (right)
|
|
if LOGO:
|
|
tg, _p, lw, lh = make_glyph(LOGO, C_CYAN, C_BG); tg.x = 10; tg.y = 9; root.append(tg)
|
|
lx = 10 + lw
|
|
else:
|
|
tg, w, h = make_text("VARASYS", FONT_M, C_CYAN, C_BG); tg.x = 10; tg.y = 8; root.append(tg)
|
|
lx = 10 + w
|
|
vtg, vw, vh = make_text("v" + APP_VERSION, FONT_S, C_DIM, C_BG); vtg.x = lx + 6; vtg.y = 8; root.append(vtg)
|
|
x = WIDTH - 12
|
|
for asset, attr in ((ICON_USB, "ic_usb_pal"), (ICON_MIDI, "ic_midi_pal")):
|
|
if asset:
|
|
tg, pal, w, h = make_glyph(asset, C_DIM, C_BG); x -= w; tg.x = x; tg.y = 8; x -= 8
|
|
root.append(tg); setattr(self, attr, pal)
|
|
root.append(rect(0, 38, WIDTH, 2, C_PANEL))
|
|
# dynamic groups
|
|
self.g_bpm = displayio.Group(); root.append(self.g_bpm) # big tempo (right)
|
|
self.g_time = displayio.Group(); root.append(self.g_time) # elapsed [of total] (left)
|
|
self.g_bar = displayio.Group(); root.append(self.g_bar) # bar [of total] (left)
|
|
self.g_train = displayio.Group(); root.append(self.g_train) # ramp / gap-trainer indicators
|
|
self.g_cont = displayio.Group(); root.append(self.g_cont) # CONT (Continue auto-advance) toggle indicator
|
|
self.g_name = displayio.Group(); root.append(self.g_name) # track title (red when edited/unsaved)
|
|
self.g_idx = displayio.Group(); root.append(self.g_idx) # set-list tab (tap to switch playlist)
|
|
self.g_grid = displayio.Group(); root.append(self.g_grid) # lanes x step pads
|
|
root.append(rect(0, LOG_TOP - 6, WIDTH, 2, C_PANEL)) # divider above the history log
|
|
self.g_log = displayio.Group(); root.append(self.g_log) # practice history (tap a row to delete)
|
|
self.g_overlay = displayio.Group(); root.append(self.g_overlay) # modal (save/revert) - drawn on top
|
|
# run/stop shows on the RGB LED; tap beats to edit, tap the title to save/revert, tap the tab to switch lists
|
|
|
|
def _place(self, group, s, x, y, fg, bg, font, right_edge=None):
|
|
while len(group): group.pop()
|
|
self.dirty = True
|
|
if not s: return
|
|
tg, w, h = make_text(s, font, fg, bg)
|
|
tg.x = (right_edge - w) if right_edge is not None else x; tg.y = y; group.append(tg)
|
|
def _center(self, group, s, cx, cy, fg, bg, font):
|
|
while len(group): group.pop()
|
|
tg, w, h = make_text(s, font, fg, bg); tg.x = cx - w//2; tg.y = cy - h//2; group.append(tg)
|
|
self.dirty = True
|
|
|
|
# ---------- program ----------
|
|
def rebuild_setlists(self):
|
|
# built-in playlists first (read-only), then user playlists from programs.json (a baked title always wins)
|
|
self.setlists = [{'title': t, 'items': it, 'builtin': True} for t, it in BUILTIN_SETLISTS]
|
|
seen = set(_slkey(t) for t, _ in BUILTIN_SETLISTS)
|
|
for t, it in load_user_setlists():
|
|
if _slkey(t) in seen: continue
|
|
seen.add(_slkey(t)); self.setlists.append({'title': t, 'items': it, 'builtin': False})
|
|
if self.sl >= len(self.setlists): self.sl = 0
|
|
def switch_setlist(self, delta=1):
|
|
if len(self.setlists) < 2: return
|
|
was = self.running
|
|
if was: self.running = False; self._log_play()
|
|
self.sl = (self.sl + delta) % len(self.setlists)
|
|
self.load(0)
|
|
if was: self.running = True; self._reset_clock(); self._start_play()
|
|
self.led_rest(); self.draw_meters()
|
|
def load(self, i):
|
|
items = self.setlists[self.sl]['items']
|
|
self.idx = i % len(items)
|
|
self.name, prog = items[self.idx]
|
|
self.bpm, self.lanes, self.bars, self.ramp, self.trainer = parse_program(prog)
|
|
self.master = self.lanes[0]; self._ramp_base = self.bpm; self._lastbar = -1; self._muted = False
|
|
self._dirty = False; self._overlay = None # fresh load -> no unsaved edits
|
|
while len(self.g_overlay): self.g_overlay.pop() # dismiss any open modal
|
|
self._reset_clock(); self.draw_bpm(); self.draw_status(); self.draw_train()
|
|
self.build_grid(); self.draw_log()
|
|
def _prog_str(self): # serialize the current (possibly edited) track to a program string
|
|
parts = ['t' + str(self.bpm)]
|
|
if self.bars: parts.append('b' + str(self.bars))
|
|
if self.ramp: parts.append('rmp%d/%d/%d' % (self.ramp.get('start', self.bpm), self.ramp['amt'], self.ramp['every']))
|
|
if self.trainer: parts.append('tr%d/%d' % (self.trainer['play'], self.trainer['mute']))
|
|
for L in self.lanes: parts.append(lane_to_str(L))
|
|
return ';'.join(parts)
|
|
|
|
# ---------- on-device editing: tap a beat to cycle it; tap the title to save/revert ----------
|
|
def _grid_hit(self, tx, ty): # map a touch to (kind, lane[, step]) on the pad grid
|
|
g = self._grid
|
|
if not g or not (g['top'] <= ty < g['top'] + g['n'] * g['rowh']): return None
|
|
li = (ty - g['top']) // g['rowh']
|
|
if li >= g['n']: return None
|
|
if tx < g['px0']: return ('lane', li) # tapped the lane label (lane editor = 0.1.0)
|
|
L = self.lanes[li]; steps = L['steps']
|
|
s = int((tx - g['px0'] - 6) * steps / g['usable'] + 0.5)
|
|
return ('beat', li, max(0, min(steps - 1, s)))
|
|
def _cycle_beat(self, li, s): # off -> normal -> accent -> ghost -> off
|
|
L = self.lanes[li]
|
|
L['levels'][s] = {0: 1, 1: 2, 2: 3, 3: 0}[L['levels'][s]]
|
|
base = self._padbase(L, s); lit = (self.lane_lit[li] == s)
|
|
self.lane_pads[li][s].color_index = base + 4 if lit else base
|
|
self._set_dirty()
|
|
def _set_dirty(self):
|
|
if not self._dirty: self._dirty = True; self.draw_status()
|
|
self.dirty = True
|
|
def toggle_continue(self):
|
|
self.continue_on = not self.continue_on; self.draw_status()
|
|
def _user_list(self, title): # find or create a user playlist
|
|
for s in self.setlists:
|
|
if not s['builtin'] and s['title'] == title: return s
|
|
s = {'title': title, 'items': [], 'builtin': False}; self.setlists.append(s); return s
|
|
def _persist_user(self): # write all user playlists back to /programs.json
|
|
user = [s for s in self.setlists if not s['builtin']]
|
|
data = {"setlists": [{"title": s['title'],
|
|
"programs": [{"name": n, "prog": p} for n, p in s['items']]} for s in user]}
|
|
try:
|
|
with open("/programs.json", "w") as f: json.dump(data, f)
|
|
return True
|
|
except OSError:
|
|
return False # editor mode: the drive is read-only to us
|
|
def _save_edit(self):
|
|
prog = self._prog_str(); sl = self.setlists[self.sl]
|
|
if sl['builtin']: # built-ins are read-only -> save a USER copy
|
|
tgt = self._user_list("My edits"); names = [n for n, _ in tgt['items']]
|
|
if self.name in names: tgt['items'][names.index(self.name)] = (self.name, prog)
|
|
else: tgt['items'].append((self.name, prog))
|
|
dest = ("My edits", self.name)
|
|
else:
|
|
sl['items'] = list(sl['items']); sl['items'][self.idx] = (self.name, prog)
|
|
dest = (sl['title'], self.name)
|
|
if not self._persist_user():
|
|
self._show_msg("Read-only: reboot without holding A"); return
|
|
self.rebuild_setlists() # refresh, then jump to the saved (user) copy
|
|
for i, s in enumerate(self.setlists):
|
|
if not s['builtin'] and s['title'] == dest[0]:
|
|
self.sl = i; names = [n for n, _ in s['items']]
|
|
self.load(names.index(dest[1]) if dest[1] in names else 0); return
|
|
self.load(0)
|
|
def _revert(self):
|
|
self.load(self.idx) # reload from source -> discard edits
|
|
# ---------- modal overlay (save / revert / message) ----------
|
|
def _show_saverevert(self):
|
|
self._overlay = 'saverevert'; g = self.g_overlay
|
|
while len(g): g.pop()
|
|
px, py, pw, ph = 24, 178, WIDTH - 48, 116
|
|
g.append(rect(px, py, pw, ph, C_PANEL)); g.append(rect(px, py, pw, 2, C_CYAN))
|
|
t, w, h = make_text("Unsaved edits", FONT_M, C_TXT, C_PANEL); t.x = px + 14; t.y = py + 12; g.append(t)
|
|
self._ovbtns = []; by = py + 44; bh = 50; gap = 12; bw = (pw - 3 * gap) // 2
|
|
for i, (lbl, col, act) in enumerate((("SAVE", C_GREEN, self._save_edit), ("REVERT", C_AMBER, self._revert))):
|
|
bx = px + gap + i * (bw + gap)
|
|
g.append(rect(bx, by, bw, bh, C_BTN)); g.append(rect(bx, by, bw, 2, col))
|
|
tt, tw, th = make_text(lbl, FONT_M, col, C_BTN); tt.x = bx + (bw - tw) // 2; tt.y = by + (bh - th) // 2; g.append(tt)
|
|
self._ovbtns.append((bx, by, bx + bw, by + bh, act))
|
|
c, cw, ch = make_text("tap outside to cancel", FONT_S, C_DIM, C_PANEL); c.x = px + 14; c.y = py + ph - 16; g.append(c)
|
|
self.dirty = True
|
|
def _show_msg(self, text):
|
|
self._overlay = 'msg'; g = self.g_overlay
|
|
while len(g): g.pop()
|
|
px, py, pw, ph = 24, 200, WIDTH - 48, 64
|
|
g.append(rect(px, py, pw, ph, C_PANEL)); g.append(rect(px, py, pw, 2, C_AMBER))
|
|
t, w, h = make_text(text[:28], FONT_S, C_TXT, C_PANEL); t.x = px + 12; t.y = py + 14; g.append(t)
|
|
t2, w2, h2 = make_text("(tap to dismiss)", FONT_S, C_DIM, C_PANEL); t2.x = px + 12; t2.y = py + 38; g.append(t2)
|
|
self.dirty = True
|
|
def _close_overlay(self):
|
|
self._overlay = None
|
|
while len(self.g_overlay): self.g_overlay.pop()
|
|
self.dirty = True
|
|
def _tap_overlay(self, tx, ty):
|
|
if self._overlay == 'msg': self._close_overlay(); return
|
|
for x0, y0, x1, y1, act in self._ovbtns:
|
|
if x0 <= tx <= x1 and y0 <= ty <= y1:
|
|
while len(self.g_overlay): self.g_overlay.pop() # clear the panel, then run the action
|
|
self._overlay = None; act(); self.dirty = True; return
|
|
self._close_overlay() # tapped outside -> cancel
|
|
def _handle_tap(self, tx, ty):
|
|
if self._overlay: self._tap_overlay(tx, ty); return
|
|
if 112 <= ty <= 126: # set-list tab line
|
|
if tx > WIDTH - 56: self.toggle_continue() # right end = CONT (auto-advance) toggle
|
|
else: self.switch_setlist(1)
|
|
return
|
|
if 128 <= ty <= 154: # track-title line
|
|
if self._dirty: self._show_saverevert()
|
|
return
|
|
hit = self._grid_hit(tx, ty)
|
|
if hit and hit[0] == 'beat': self._cycle_beat(hit[1], hit[2]); return
|
|
self._tap_log(tx, ty) # else the practice log
|
|
def _step_dur(self, L, step):
|
|
beat = 60_000_000_000 / self.bpm
|
|
if L['poly']: # ~ polymeter: fit this lane's whole cycle into lane 1's bar
|
|
m = self.lanes[0]; master_bar = beat * (m['steps'] // m['sub'])
|
|
return int(master_bar / L['steps'])
|
|
sub = L['sub']
|
|
if L['swing'] and sub % 2 == 0: # swing even subdivisions: long-short (2:1) pairs
|
|
pair = beat / (sub // 2)
|
|
return int(pair * 2 / 3) if (step % sub) % 2 == 0 else int(pair / 3)
|
|
return int(beat / sub) # straight: a step = one beat / subdivision
|
|
def _reset_clock(self):
|
|
now = time.monotonic_ns()
|
|
for L in self.lanes:
|
|
L['next'] = now; L['step'] = -1
|
|
self._m_steps = 0 # restart the bar count
|
|
self._seg_start = time.monotonic() # and the on-screen timer (resets with the bar counter)
|
|
|
|
# ---------- audio + light ----------
|
|
def click(self, level):
|
|
self.buz.frequency = {2: 2300, 1: 1600, 3: 1050}.get(level, 1600)
|
|
self.buz.duty_cycle = {2: 42000, 1: 30000, 3: 14000}.get(level, 30000)
|
|
self.buz_off = time.monotonic_ns() + 22_000_000
|
|
def _led_base(self):
|
|
return LED_RUN if self.running else LED_IDLE # dim red while playing / dim green when stopped
|
|
def flash(self, level):
|
|
self.rgb = LEVEL_RGB.get(level, (0, 150, 255)) # bright beat pulse, fades back to the base in tick()
|
|
self.led.set(*self.rgb)
|
|
def led_rest(self): # settle to the resting colour (green idle / red running)
|
|
self.rgb = self._led_base()
|
|
self.led.set(*self.rgb)
|
|
def midi_send(self, note, vel): # device-as-conductor: a note per click to the computer
|
|
if self.midi is None: return
|
|
try: self.midi.write(bytes([0x90, note, vel])) # Note On (percussive - no Note Off needed)
|
|
except Exception: pass
|
|
|
|
# ---------- transport ----------
|
|
def toggle(self):
|
|
self.running = not self.running
|
|
if self.running: self._reset_clock(); self._start_play()
|
|
else: self.buz.duty_cycle = 0; self.reset_playheads(); self._log_play()
|
|
self.led_rest(); self.draw_meters() # LED shows run state: red running / green stopped
|
|
def set_bpm(self, v):
|
|
v = max(30, min(300, v))
|
|
if v != self.bpm:
|
|
self.bpm = v
|
|
self.draw_bpm(); self.draw_meters() # total time depends on bpm
|
|
def goto(self, i):
|
|
was = self.running
|
|
if was: self.running = False; self._log_play() # close out the track that was playing
|
|
self.load(i)
|
|
if was: self.running = True; self._reset_clock(); self._start_play()
|
|
self.led_rest(); self.draw_meters()
|
|
def tap(self):
|
|
now = time.monotonic()
|
|
if not hasattr(self, '_taps'): self._taps = []
|
|
self._taps = [t for t in self._taps if now - t < 2.4]
|
|
self._taps.append(now)
|
|
if len(self._taps) >= 2:
|
|
span = (self._taps[-1] - self._taps[0]) / (len(self._taps) - 1)
|
|
if span > 0: self.set_bpm(round(60 / span))
|
|
|
|
# ---------- scheduler ----------
|
|
def tick(self):
|
|
now = time.monotonic_ns()
|
|
if self.buz_off and now >= self.buz_off: self.buz.duty_cycle = 0; self.buz_off = 0
|
|
if self.running:
|
|
fired = []
|
|
for li, L in enumerate(self.lanes):
|
|
adv = False
|
|
while now >= L['next']:
|
|
L['step'] = (L['step'] + 1) % L['steps']
|
|
if li == 0:
|
|
self._m_steps += 1 # count master-lane steps -> bars
|
|
nb = self._m_steps // L['steps']
|
|
if nb != self._lastbar: self._lastbar = nb; self._on_new_bar(nb) # ramp + gap-trainer
|
|
lvl = 0 if L['mute'] else L['levels'][L['step']]
|
|
if lvl > 0:
|
|
fired.append(lvl)
|
|
if not self._muted: # gap trainer: silent during the rest bars
|
|
self.midi_send(SOUND_GM.get(L['sound'], GM_DEFAULT), MIDI_VEL.get(lvl, 90))
|
|
L['next'] += self._step_dur(L, L['step']); adv = True
|
|
if adv and li < len(self.lane_pads): self._move_playhead(li, L['step'])
|
|
if fired and not self._muted:
|
|
best = max(fired, key=lambda l: PRIO.get(l, 0))
|
|
if not MUTE_BUZZER and not self.midi_host: self.click(best) # computer plays it instead
|
|
self.flash(best)
|
|
base = self._led_base() # decay the beat pulse back down to the red running base
|
|
if self.rgb != base:
|
|
r = base[0] + (self.rgb[0]-base[0])*7//10
|
|
g = base[1] + (self.rgb[1]-base[1])*7//10
|
|
b = base[2] + (self.rgb[2]-base[2])*7//10
|
|
if abs(r-base[0])+abs(g-base[1])+abs(b-base[2]) < 6: r, g, b = base
|
|
self.rgb = (r, g, b); self.led.set(r, g, b)
|
|
if self._advance: # Continue: roll to the next item at the segment end
|
|
self._advance = False
|
|
self.load((self.idx + 1) % len(self.setlists[self.sl]['items'])); self.led_rest()
|
|
def _on_new_bar(self, bar):
|
|
if self.bars and bar > 0 and bar % self.bars == 0: # segment boundary
|
|
self._seg_start = time.monotonic() # timer resets with the bar counter
|
|
if self.ramp: self.set_bpm(self._ramp_base) # ramp restarts each segment
|
|
if self.continue_on: self._advance = True # Continue: roll to the next item
|
|
elif self.ramp and bar > 0 and bar % self.ramp['every'] == 0:
|
|
self.set_bpm(self.bpm + self.ramp['amt']) # mid-segment ramp step
|
|
t = self.trainer # gap trainer: silence during the rest bars
|
|
self._muted = bool(t and (t['play'] + t['mute']) and (bar % (t['play'] + t['mute'])) >= t['play'])
|
|
|
|
# ---------- inputs ----------
|
|
def poll(self):
|
|
a = self.btnA.value
|
|
if (not a) and self._aPrev: self.toggle()
|
|
self._aPrev = a
|
|
b = self.btnB.value
|
|
if (not b) and self._bPrev: self.tap()
|
|
self._bPrev = b
|
|
now = time.monotonic_ns()
|
|
if now >= self._joyNext:
|
|
x = self.jx.value - 32768; y = self.jy.value - 32768
|
|
if JOY_INVERT_X: x = -x
|
|
if JOY_INVERT_Y: y = -y
|
|
if abs(y) > JOY_DEADZONE:
|
|
self.set_bpm(self.bpm + (1 if y > 0 else -1) * (5 if abs(y) > 26000 else 1))
|
|
self._joyNext = now + 70_000_000
|
|
elif abs(x) > JOY_DEADZONE:
|
|
self.goto(self.idx + (1 if x > 0 else -1)); self._joyNext = now + 350_000_000; return
|
|
else:
|
|
self._joyNext = now + 20_000_000
|
|
nowms = time.monotonic()
|
|
if nowms >= self._touchNext: # poll touch ~30x/s (the I2C read adds loop latency -> MIDI jitter)
|
|
self._touchNext = nowms + 0.033
|
|
pt = self.touch.read()
|
|
if pt:
|
|
self._touchSeen = nowms
|
|
if not self._touchDown:
|
|
self._touchDown = True; self._handle_tap(pt[0], pt[1])
|
|
elif self._touchDown and (nowms - self._touchSeen) > 0.14:
|
|
self._touchDown = False
|
|
# USB-MIDI in: any byte = a host is listening (heartbeat); also assemble SysEx (clock / pushed programs)
|
|
if self.midi_in is not None:
|
|
try: n = self.midi_in.readinto(self._mbuf)
|
|
except Exception: n = 0
|
|
if n:
|
|
self.last_midi_in = nowms
|
|
self._feed_midi(self._mbuf, n)
|
|
host = bool(self.last_midi_in) and (nowms - self.last_midi_in) < 1.0
|
|
if host != self.midi_host:
|
|
self.midi_host = host
|
|
if host: self.buz.duty_cycle = 0 # silence the buzzer when the computer takes over
|
|
self.led_rest(); self.draw_icons()
|
|
uc = bool(getattr(supervisor.runtime, "usb_connected", True)) # connected to a computer?
|
|
if uc != self.usb_conn:
|
|
self.usb_conn = uc; self.draw_icons()
|
|
|
|
# ---------- drawing ----------
|
|
def draw_bpm(self):
|
|
self._place(self.g_bpm, str(self.bpm), 0, 44, C_TXT, C_BG, FONT_L, right_edge=WIDTH-12)
|
|
def draw_status(self): # set-list tab (tap=switch) + CONT toggle, above the item title
|
|
sl = self.setlists[self.sl]
|
|
# tab: playlist + position; muted = built-in (read-only), cyan = your own
|
|
self._place(self.g_idx, "%s %d/%d" % (sl['title'][:11], self.idx + 1, len(sl['items'])),
|
|
12, 118, C_MUTE if sl['builtin'] else C_CYAN, C_BG, FONT_S)
|
|
self._place(self.g_cont, "CONT", 0, 118, C_GREEN if self.continue_on else C_DIM, C_BG, FONT_S, right_edge=WIDTH-12)
|
|
# title turns red when edited (tap it to save/revert)
|
|
self._place(self.g_name, self.name[:20], 12, 134, C_RED if self._dirty else C_TXT, C_BG, FONT_M)
|
|
def draw_train(self): # ramp + gap-trainer indicators (symbol + params), when set
|
|
g = self.g_train
|
|
while len(g): g.pop()
|
|
x = 12; y = 100
|
|
if self.ramp:
|
|
up = self.ramp['amt'] >= 0
|
|
pts = [(0, 9), (12, 9), (12, 0)] if up else [(0, 0), (0, 9), (12, 9)] # rising / falling ramp
|
|
g.append(vectorio.Polygon(pixel_shader=solid(C_AMBER), points=pts, x=x, y=y)); x += 16
|
|
a = self.ramp['amt']; lbl = ("+%d" % a if a >= 0 else "%d" % a) + "/%db" % self.ramp['every']
|
|
tg, w, h = make_text(lbl, FONT_S, C_AMBER, C_BG); tg.x = x; tg.y = y; g.append(tg); x += w + 14
|
|
if self.trainer:
|
|
g.append(rect(x, y, 4, 9, C_CYAN)); g.append(rect(x + 6, y, 4, 9, C_DIM)) # play | rest
|
|
x += 14
|
|
tg, w, h = make_text("%d/%db" % (self.trainer['play'], self.trainer['mute']), FONT_S, C_CYAN, C_BG)
|
|
tg.x = x; tg.y = y; g.append(tg)
|
|
self.dirty = True
|
|
def draw_icons(self): # recolor the MIDI/USB icons by state (tear-free palette swap)
|
|
if self.ic_midi_pal is not None:
|
|
_recolor(self.ic_midi_pal, C_GREEN if self.midi_host else C_DIM, C_BG)
|
|
if self.ic_usb_pal is not None:
|
|
_recolor(self.ic_usb_pal, C_CYAN if self.usb_conn else C_DIM, C_BG)
|
|
self.dirty = True
|
|
def _fmt_t(self, s): # m:ss, or h:mm:ss past an hour
|
|
s = int(s)
|
|
return "%d:%02d:%02d" % (s // 3600, (s % 3600) // 60, s % 60) if s >= 3600 else "%d:%02d" % (s // 60, s % 60)
|
|
def draw_meters(self): # running time [of total] + bar [of total]; ~4x/s from run()
|
|
run = self.running and self.play_start is not None
|
|
mlen = self.lanes[0]['steps'] if self.lanes else 1
|
|
bpb = (self.lanes[0]['steps'] // max(1, self.lanes[0]['sub'])) if self.lanes else 4
|
|
el = (time.monotonic() - self._seg_start) if run else 0 # time within the current segment (resets with the bar)
|
|
mbars = self._m_steps // max(1, mlen) # whole master bars elapsed
|
|
cur = ("%d" % ((mbars % self.bars + 1) if self.bars else (mbars + 1))) if run else "-" # cycle 1..N
|
|
if self.bars: # track has a length (b<n>): show "X of TOTAL"
|
|
ts = "%s of %s" % (self._fmt_t(el), self._fmt_t(self.bars * bpb * 60.0 / self.bpm))
|
|
bs = "bar %s of %d" % (cur, self.bars)
|
|
else:
|
|
ts = self._fmt_t(el); bs = "bar %s" % cur
|
|
if ts != self._lastTs:
|
|
self._place(self.g_time, ts, 12, 50, C_TXT, C_BG, FONT_M); self._lastTs = ts
|
|
if bs != self._lastBs:
|
|
self._place(self.g_bar, bs, 12, 78, C_MUTE, C_BG, FONT_M); self._lastBs = bs
|
|
|
|
# ---------- pad grid (each lane = a row of step pads; playhead lit as it plays) ----------
|
|
def _padbase(self, L, s):
|
|
return 0 if L['mute'] else L['levels'][s]
|
|
def build_grid(self):
|
|
while len(self.g_grid): self.g_grid.pop()
|
|
self.lane_pads = []; self.lane_lit = []
|
|
n = min(len(self.lanes), MAXLANES)
|
|
top = GRID_TOP; rowh = min(40, ((LOG_TOP - 10) - top) // max(1, n))
|
|
px0 = 60; usable = WIDTH - 8 - px0 - 12; gridh = n * rowh
|
|
self._grid = {'top': top, 'rowh': rowh, 'px0': px0, 'usable': usable, 'n': n} # for touch hit-testing
|
|
# vertical gridlines at the master lane's beats, full height -> beats line up across lanes
|
|
m = self.lanes[0]; mbeats = max(1, m['steps'] // max(1, m['sub']))
|
|
for bcol in range(mbeats):
|
|
self.g_grid.append(rect(px0 + 6 + (bcol * usable) // mbeats, top, 1, gridh, C_GRID))
|
|
for li in range(n):
|
|
L = self.lanes[li]; y = top + li * rowh; cy = y + rowh // 2
|
|
tg, w, h = make_text((L.get('sound', '') or '?')[:7], FONT_S, C_MUTE, C_BG)
|
|
tg.x = 8; tg.y = cy - h // 2; self.g_grid.append(tg)
|
|
steps = L['steps']; sub = L['sub']; stepw = max(1, usable // steps)
|
|
side = max(5, min(15, stepw - 1, rowh - 6)) # square edge for the main pulse
|
|
rad = max(2, min(side // 2, stepw // 2 - 1)) # smaller circle for subdivisions
|
|
pads = []
|
|
for s in range(steps):
|
|
cxp = px0 + 6 + (s * usable) // steps # proportional -> beats line up across lanes
|
|
if s % sub == 0: # main beat -> square
|
|
p = vectorio.Rectangle(pixel_shader=self.pad_pal, width=side, height=side,
|
|
x=cxp - side // 2, y=cy - side // 2)
|
|
else: # subdivision -> circle
|
|
p = vectorio.Circle(pixel_shader=self.pad_pal, radius=rad, x=cxp, y=cy)
|
|
p.color_index = self._padbase(L, s); self.g_grid.append(p); pads.append(p)
|
|
self.lane_pads.append(pads); self.lane_lit.append(-1)
|
|
self.dirty = True
|
|
def _move_playhead(self, li, step):
|
|
pads = self.lane_pads[li]; prev = self.lane_lit[li]
|
|
if 0 <= prev < len(pads): pads[prev].color_index = self._padbase(self.lanes[li], prev)
|
|
if step < len(pads): pads[step].color_index = self._padbase(self.lanes[li], step) + 4
|
|
self.lane_lit[li] = step; self.dirty = True
|
|
def reset_playheads(self):
|
|
for li, pads in enumerate(self.lane_pads):
|
|
prev = self.lane_lit[li]
|
|
if 0 <= prev < len(pads): pads[prev].color_index = self._padbase(self.lanes[li], prev)
|
|
self.lane_lit[li] = -1
|
|
self.dirty = True
|
|
|
|
# ---------- practice history (saved to /history.json, next to programs.json) ----------
|
|
def _probe_write(self):
|
|
try:
|
|
with open("/.wtest", "w") as f: f.write("1")
|
|
try: os.remove("/.wtest")
|
|
except Exception: pass
|
|
return True
|
|
except OSError:
|
|
return False # editor mode: the computer owns the FS
|
|
def _load_log(self):
|
|
try:
|
|
with open("/history.json") as f: return json.load(f).get("log", [])
|
|
except Exception:
|
|
return []
|
|
def _save_log(self):
|
|
if not self.can_write: return
|
|
try:
|
|
with open("/history.json", "w") as f: json.dump({"log": self.log[:200]}, f)
|
|
except OSError:
|
|
self.can_write = False
|
|
def _start_play(self):
|
|
self.play_start = time.monotonic(); self.play_bpm = self.bpm; self.play_name = self.name
|
|
def _log_play(self):
|
|
if self.play_start is None: return
|
|
dur = int(time.monotonic() - self.play_start); self.play_start = None
|
|
if dur < MIN_LOG_SEC: return # skip plays under 5 seconds
|
|
mlen = self.lanes[0]['steps'] if self.lanes else 1
|
|
t = time.localtime()
|
|
self.log.insert(0, {"t": "%02d:%02d" % (t.tm_hour, t.tm_min), "bpm": self.play_bpm,
|
|
"dur": dur, "bars": self._m_steps // max(1, mlen), "name": self.play_name})
|
|
del self.log[200:]; self._armed = None
|
|
self._save_log(); self.draw_log()
|
|
def draw_log(self):
|
|
g = self.g_log
|
|
while len(g): g.pop()
|
|
self.log_rows = []
|
|
hdr, w, h = make_text("PRACTICE LOG - THIS TRACK", FONT_S, C_MUTE, C_BG); hdr.x = 10; hdr.y = LOG_TOP; g.append(hdr)
|
|
rows = [(i, e) for i, e in enumerate(self.log) if e.get("name") == self.name] # current track only
|
|
if not rows:
|
|
tg, w, h = make_text("no plays over 5s yet", FONT_S, C_DIM, C_BG); tg.x = 10; tg.y = LOG_TOP + LOG_ROWH; g.append(tg)
|
|
self.dirty = True; return
|
|
y = LOG_TOP + LOG_ROWH + 2
|
|
for k in range(min(LOG_ROWS, len(rows))):
|
|
oi, e = rows[k]; armed = (oi == self._armed) # oi = index into self.log (for delete)
|
|
dur = "%d:%02d" % (e["dur"] // 60, e["dur"] % 60)
|
|
bars = e.get("bars", 0); bstr = (" %dbar" % bars) if bars else ""
|
|
line = "%s%s %3dbpm %s%s" % ("x " if armed else "", e.get("t", "--:--"), e["bpm"], dur, bstr)
|
|
tg, w, h = make_text(line, FONT_S, C_AMBER if armed else C_TXT, C_BG); tg.x = 10; tg.y = y; g.append(tg)
|
|
self.log_rows.append((y - 2, y + LOG_ROWH - 2, oi))
|
|
y += LOG_ROWH
|
|
self.dirty = True
|
|
def _tap_log(self, x, ty):
|
|
for y0, y1, idx in self.log_rows:
|
|
if y0 <= ty <= y1:
|
|
if self._armed == idx: del self.log[idx]; self._armed = None; self._save_log(); self.draw_log() # confirm delete
|
|
else: self._armed = idx; self.draw_log() # arm (tap again)
|
|
return
|
|
if self._armed is not None: self._armed = None; self.draw_log() # tapped elsewhere -> cancel
|
|
|
|
# ---------- USB-MIDI in: SysEx assembler (clock + editor-pushed programs) ----------
|
|
def _feed_midi(self, buf, n):
|
|
for i in range(n):
|
|
b = buf[i]
|
|
if b == 0xF0: self._sx = bytearray(); self._sxon = True
|
|
elif b == 0xF7:
|
|
if self._sxon: self._handle_sysex(self._sx)
|
|
self._sxon = False
|
|
elif b >= 0xF8: pass # real-time (e.g. Active Sensing 0xFE) - ignore
|
|
elif self._sxon:
|
|
if len(self._sx) < 60000: self._sx.append(b) # big enough for a pushed firmware (app.py)
|
|
else: self._sxon = False # overflow guard
|
|
def _handle_sysex(self, sx):
|
|
if len(sx) < 2 or sx[0] != 0x7D: return # 0x7D = our (educational) manufacturer id
|
|
cmd = sx[1]
|
|
if cmd == 0x01 and len(sx) >= 8 and rtc is not None: # set clock: yr-2000, mo, dd, hh, mm, ss
|
|
try: rtc.RTC().datetime = time.struct_time((2000 + sx[2], sx[3], sx[4], sx[5], sx[6], sx[7], 0, -1, -1))
|
|
except Exception: pass
|
|
elif cmd == 0x02: # version query -> reply 0x03 + APP_VERSION
|
|
if self.midi: self.midi.write(bytes([0xF0, 0x7D, 0x03]) + APP_VERSION.encode() + bytes([0xF7]))
|
|
elif cmd == 0x10: # write /programs.json (user playlists) pushed from the editor
|
|
try:
|
|
with open("/programs.json", "wb") as f: f.write(bytes(sx[2:]))
|
|
self.rebuild_setlists(); self.load(0) # built-ins untouched; show the refreshed lists
|
|
self._ack(True)
|
|
except Exception:
|
|
self._ack(False) # read-only (editor mode) etc.
|
|
# A/B firmware update, sent as small flow-controlled chunks (a single huge SysEx overruns the
|
|
# USB-MIDI input buffer and arrives corrupt). begin(0x21,len) -> data(0x22)* -> commit(0x23).
|
|
elif cmd == 0x21: # BEGIN firmware transfer: open the .mpy staging file
|
|
try:
|
|
try: self._fw.close()
|
|
except Exception: pass
|
|
self._fw = open("/app.new", "wb"); self._ack(True)
|
|
except Exception: # read-only (editor mode) / no space
|
|
self._fw = None; self._ack(False)
|
|
elif cmd == 0x22: # DATA: a base64 chunk (multiple of 4) -> decode -> append
|
|
try:
|
|
if self._fw is None or a2b_base64 is None: raise OSError()
|
|
self._fw.write(a2b_base64(bytes(sx[2:]))); self._ack(True)
|
|
except Exception:
|
|
try: self._fw.close()
|
|
except Exception: pass
|
|
self._fw = None; self._ack(False)
|
|
elif cmd == 0x23: # COMMIT: verify it's a CircuitPython .mpy, then A/B install
|
|
try:
|
|
try: self._fw.close()
|
|
except Exception: pass
|
|
self._fw = None; gc.collect()
|
|
with open("/app.new", "rb") as f: head = f.read(2)
|
|
if os.stat("/app.new")[6] < 4000 or len(head) < 2 or head[0] != 0x43 or head[1] != 0x06:
|
|
try: os.remove("/app.new") # not a CircuitPython mpy v6 -> reject, keep the working build
|
|
except OSError: pass
|
|
self._ack(False); return
|
|
try: os.remove("/app.bak")
|
|
except OSError: pass
|
|
os.rename("/app.mpy", "/app.bak") # current build becomes the rollback
|
|
os.rename("/app.new", "/app.mpy")
|
|
open("/trial", "w").close() # arm the trial; the loader reverts if it won't boot
|
|
self._ack(True); time.sleep(0.4); supervisor.reload()
|
|
except Exception: # catch ALL (read-only, MemoryError, ...) -> never brick
|
|
self._ack(False)
|
|
def _ack(self, ok):
|
|
if self.midi: self.midi.write(bytes([0xF0, 0x7D, 0x7F if ok else 0x7E, 0xF7]))
|
|
|
|
def run(self):
|
|
if self.touch.addr is None:
|
|
print("GT911 touch not found")
|
|
boot = time.monotonic()
|
|
try: os.stat("/trial"); committed = False # we're a freshly-pushed build on trial
|
|
except OSError: committed = True
|
|
while True:
|
|
self.tick(); self.poll()
|
|
tnow = time.monotonic()
|
|
if tnow >= self._uiNext: # ~4x/s: tick the stopwatch + bar counter
|
|
self._uiNext = tnow + 0.25; self.draw_meters()
|
|
if not committed and tnow - boot > 5: # booted & ran fine for 5s -> confirm the update
|
|
try: os.remove("/trial")
|
|
except Exception: pass
|
|
committed = True
|
|
# Refresh at most ~30x/s. display.refresh() BLOCKS while it streams pixels over SPI, which
|
|
# would otherwise delay the next beat's MIDI note and make the audio stutter; throttling it
|
|
# keeps the click timing tight (the visuals lag a few ms, which is imperceptible).
|
|
if self.dirty and tnow >= self._refreshNext:
|
|
if self.display.refresh(): self.dirty = False
|
|
self._refreshNext = tnow + 0.033
|
|
time.sleep(0.0005)
|
|
|
|
App().run()
|