VARASYS/metronome
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

PM_K-1 0.0.20: per-pad chunking + step grids + smarter refresh

Browse source 
Three changes; one diagnosis: the 0.0.19 freeze + stutter were both symptoms of the
predictive refresh skip being too aggressive at fast subdivisions (scanning ALL lanes
+ a 35ms window meant the next beat is *always* within the skip window at fine
subdivisions, so the only refresh was the 0.5s force-after fallback, audible as both
a flickery freeze AND a periodic 30ms stutter).

PER-PAD CHUNKING (your ask):
- _grid_rebuild_step now builds ONE rectangle per loop iter (was one whole lane row).
- The first chunk on a lane draws the instrument label + caches the lane's geometry;
  subsequent chunks each emit one pad rect/circle.
- Each chunk is ~5-10ms instead of ~30-50ms, so tick() interleaves at sub-beat
  resolution and the grid fills in pad-by-pad over ~600ms after the seam without
  ever blocking long enough to miss a beat at any reasonable BPM.

STEP GRIDS:
- L['durs'] is a tuple of int ns per step, precomputed once per lane.
- tick()'s inner-loop scheduler is now L['next'] += L['durs'][L['step']] -- a tuple
  index, no method call, no dict lookups, no division.
- _rebuild_dur(L) and _rebuild_dur_all() wired into every bpm-change path:
  set_bpm / load / _do_advance / _prepare_next / _slave_tick (Clock In) / _lane_dirty
  (the lane editor; rebuilds master-affects-poly cascade if lane 0 changes structurally),
  + the per-master-step ramp interpolator. The synchronous _step_dur stays as a fallback
  for paths the inner loop doesn't hit (e.g. start-of-segment calculations elsewhere).

SMARTER REFRESH:
- Master-lane-only window (other lanes are subdivisions; a few ms of jitter there is
  imperceptible musically).
- 10ms window (down from 35) so refresh runs in the gap between master beats even at
  fast subdivisions.
- 20Hz throttle (down from 30) so the per-refresh blocking budget is bigger.
- 200ms force fallback (down from 500) so visuals + titles stay live -- this is the
  fix for "the track titles don't get reliably updated."

DEFER set_bpm DRAWS:
- set_bpm no longer calls draw_bpm/draw_meters; the 4Hz UI tick already redraws them.
- Joystick spam used to allocate a fresh text bitmap per nudge -> fragmentation
  pressure + potential GC stutter.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Me Here 2026-05-30 08:58:13 -05:00
parent 672f892ea1
commit 09144c9892
1 changed files with 81 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

119
pico-cp/app.py
View file

@ -18,7 +18,7 @@
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.19" # firmware version (the A/B updater pushes/compares this)
APP_VERSION = "0.0.20" # 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:
@ -453,7 +453,8 @@ class App:
self.continue_on = False; self._advance = False; self._grid = {} # auto-advance + pad hit-test geometry
self._next_pending = None; self._seam_t = 0; self._need_redraw = False # gapless seam between tracks
self._heavy_redraw_at = 0 # deferred build_grid + draw_log deadline (so B's intro isn't blocked by SPI/alloc)
self._grid_li = None; self._grid_n = 0; self._grid_geo = (0, 0, 0, 0) # chunked build_grid progress (1 lane / loop iter)
self._grid_li = None; self._grid_n = 0; self._grid_geo = (0, 0, 0, 0) # chunked build_grid progress (1 PAD / loop iter)
self._grid_pi = 0; self._grid_lane_st = None; self._grid_pads = [] # per-lane sub-state for sub-pad chunking
self._heavy_log_pending = False
self._beat_ns = 60_000_000_000 // self.bpm # cached: ns per quarter note; refreshed on every bpm change
self._note_buf = bytearray([0x90, 0, 0]) # reused for every Note On (no per-click bytes() alloc)
@ -556,6 +557,7 @@ class App:
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._beat_ns = 60_000_000_000 // max(1, self.bpm); self._rebuild_dur_all() # step grids ready for this lane set
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
self._next_pending = None; self._need_redraw = False # discard any prepared seam (user navigated away)
@ -721,6 +723,8 @@ class App:
L['levels'] = [(2 if (i // sub) in starts else 1) if i % sub == 0 else 0 for i in range(L['steps'])]
def _lane_dirty(self, structural):
if structural: self._regen_levels(self.lanes[self._edit_li])
if structural and self._edit_li == 0: self._rebuild_dur_all() # master changed -> polymeter lanes follow
else: self._rebuild_dur(self.lanes[self._edit_li])
self.build_grid()
if not self._dirty: self._dirty = True; self.draw_status()
self._draw_laneedit() # refresh the modal with the new values
@ -947,16 +951,29 @@ class App:
with open("/settings.json", "w") as f: json.dump(d, f)
except OSError: self.can_write = False
def _step_dur(self, L, step):
beat = self._beat_ns # cached ns/beat (refreshed on every bpm change + ramp tick)
if L['poly']: # ~ polymeter: fit this lane's whole cycle into lane 1's bar
def _step_dur(self, L, step): # legacy fallback (still used by _start_play tap-tempo path)
beat = self._beat_ns
if L['poly']:
m = self.lanes[0]; master_bar = beat * (m['steps'] // m['sub'])
return master_bar // L['steps']
sub = L['sub']
if L['swing'] and sub % 2 == 0: # swing even subdivisions: long-short (2:1) pairs
if L['swing'] and sub % 2 == 0:
pair = beat // (sub // 2)
return (pair * 2) // 3 if (step % sub) % 2 == 0 else pair // 3
return beat // sub # straight: a step = one beat / subdivision
return beat // sub
def _rebuild_dur(self, L): # cache the per-step ns durations into L['durs'] (tuple lookup is ~10us)
beat = self._beat_ns
sub = max(1, L['sub']); steps = max(1, L['steps'])
if L.get('poly') and self.lanes: # polymeter: spread this lane's cycle across master's bar
m = self.lanes[0]; master_bar = beat * (m['steps'] // max(1, m['sub']))
d = master_bar // steps; L['durs'] = tuple(d for _ in range(steps))
elif L.get('swing') and sub % 2 == 0: # swing: long-short pairs
pair = beat // max(1, sub // 2); lng = (pair * 2) // 3; sht = pair // 3
L['durs'] = tuple(lng if (s % sub) % 2 == 0 else sht for s in range(steps))
else: # straight: every step is beat/sub long
d = beat // sub; L['durs'] = tuple(d for _ in range(steps))
def _rebuild_dur_all(self): # called on bpm change + lane mutation + track swap
for L in self.lanes: self._rebuild_dur(L)
def _reset_clock(self):
now = time.monotonic_ns()
for L in self.lanes:
@ -1003,8 +1020,9 @@ class App:
def set_bpm(self, v):
v = max(5, min(300, v))
if v != self.bpm:
self.bpm = v; self._beat_ns = 60_000_000_000 // v # keep cached beat duration in sync
self.draw_bpm(); self.draw_meters() # total time depends on bpm
self.bpm = v; self._beat_ns = 60_000_000_000 // v
self._rebuild_dur_all() # step grids follow the new beat duration
# Don't draw here -- the 4Hz UI tick redraws bpm/meters; calling per joystick nudge allocated text bitmaps fast enough to trigger GC pauses
def goto(self, i):
was = self.running
if was: self.running = False; self._log_play() # close out the track that was playing
@ -1043,14 +1061,16 @@ class App:
bar_pos = self._m_steps / mlen
seg_bar = (bar_pos % self.bars) if self.bars else bar_pos
new_bpm = max(5, min(300, int(self._ramp_base + seg_bar / self.ramp['every'] * self.ramp['amt'])))
if new_bpm != self.bpm: self.bpm = new_bpm; self._beat_ns = 60_000_000_000 // new_bpm
if new_bpm != self.bpm:
self.bpm = new_bpm; self._beat_ns = 60_000_000_000 // new_bpm
self._rebuild_dur_all() # ramp moves bpm -> step grids follow
lvl = 0 if L['mute'] else L['levels'][L['step']]
if lvl > 0:
p = PRIO.get(lvl, 0)
if p > fired_prio: fired_prio = p; fired_best = lvl # accent > normal > ghost
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
L['next'] += L['durs'][L['step']]; adv = True # zero method call, zero dict lookup, just a tuple index
if adv and li < len(self.lane_pads): self._move_playhead(li, L['step'])
if fired_best and not self._muted:
if not MUTE_SPEAKER and not (SPEAKER_AUTO_MUTE and self.midi_host):
@ -1098,6 +1118,17 @@ class App:
bpm, lanes, bars, ramp, trainer = parse_program(prog)
except MemoryError:
gc.collect(); return # leave _next_pending None -> the segment just loops
beat = 60_000_000_000 // max(1, bpm) # pre-compute B's durs against B's bpm so the seam swap is allocation-free
for L in lanes:
sub = max(1, L['sub']); steps = max(1, L['steps'])
if L.get('poly'):
m = lanes[0]; mbar = beat * (m['steps'] // max(1, m['sub']))
d = mbar // steps; L['durs'] = tuple(d for _ in range(steps))
elif L.get('swing') and sub % 2 == 0:
pair = beat // max(1, sub // 2); lng = (pair * 2) // 3; sht = pair // 3
L['durs'] = tuple(lng if (s % sub) % 2 == 0 else sht for s in range(steps))
else:
d = beat // sub; L['durs'] = tuple(d for _ in range(steps))
self._next_pending = {'lanes': lanes, 'bpm': bpm, 'bars': bars, 'ramp': ramp,
'trainer': trainer, 'name': name, 'idx': nxt}
def _do_advance(self): # gapless seam: swap the prepared track in at seam_t
@ -1106,6 +1137,7 @@ class App:
self._next_pending = None
self.lanes = n['lanes']; self.bpm = n['bpm']; self.bars = n['bars']
self.ramp = n['ramp']; self.trainer = n['trainer']; self.name = n['name']; self.idx = n['idx']
self._beat_ns = 60_000_000_000 // max(1, self.bpm); self._rebuild_dur_all() # B's step grids built at the seam
self._ramp_base = self.bpm; self._lastbar = -1; self._muted = False; self._m_steps = 0
self._dirty = False; self._overlay = None
while len(self.g_overlay): self.g_overlay.pop()
@ -1239,29 +1271,40 @@ class App:
self._grid_n = n
self._grid_geo = (top, rowh, px0, usable)
self._grid_li = 0 if n > 0 else None
self._grid_pi = 0; self._grid_lane_st = None; self._grid_pads = [] # start at lane 0, pad 0, no lane initialized yet
self.dirty = True
def _grid_rebuild_step(self): # build ONE lane row; sets _grid_li=None when done
def _grid_rebuild_step(self): # PER-PAD chunk: build at most one rectangle, then yield
li = self._grid_li
if li is None: return
if li >= self._grid_n or li >= len(self.lanes):
self._grid_li = None; return # done; main loop falls through to draw_log
self._grid_li = None; return # whole rebuild done -> main loop runs draw_log
L = self.lanes[li]
top, rowh, px0, usable = self._grid_geo
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
pal = self.pad_pal; pads = []
for s in range(steps):
cxp = px0 + 6 + (s * usable) // steps # proportional -> beats line up across lanes
if s % sub == 0:
p = vectorio.Rectangle(pixel_shader=pal, width=side, height=side, x=cxp - side // 2, y=cy - side // 2)
else:
p = vectorio.Circle(pixel_shader=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._grid_li = li + 1
y = top + li * rowh; cy = y + rowh // 2
st = self._grid_lane_st
if st is None: # first chunk on this lane: draw the instrument label + cache the geometry
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
self._grid_lane_st = (cy, steps, sub, stepw, side, rad)
self._grid_pi = 0; self._grid_pads = []; self.dirty = True
return # one chunk = "init this lane"; next iter does the first pad
cy_, steps, sub, stepw, side, rad = st
s = self._grid_pi
if s >= steps: # this lane finished; commit and advance to next
self.lane_pads.append(self._grid_pads); self.lane_lit.append(-1)
self._grid_pads = []; self._grid_lane_st = None; self._grid_li = li + 1
return
cxp = px0 + 6 + (s * usable) // steps # proportional -> beats line up across lanes
pal = self.pad_pal
if s % sub == 0:
p = vectorio.Rectangle(pixel_shader=pal, width=side, height=side, x=cxp - side // 2, y=cy_ - side // 2)
else:
p = vectorio.Circle(pixel_shader=pal, radius=rad, x=cxp, y=cy_)
p.color_index = self._padbase(L, s); self.g_grid.append(p); self._grid_pads.append(p)
self._grid_pi = s + 1
self.dirty = True
def _move_playhead(self, li, step):
pads = self.lane_pads[li]; prev = self.lane_lit[li]
@ -1362,7 +1405,8 @@ class App:
if self._clock_in_avg == 0: self._clock_in_avg = interval
else: self._clock_in_avg = (self._clock_in_avg * 7 + interval) // 8 # exponential smoothing, alpha = 1/8
new_bpm = max(5, min(300, int(60_000_000_000 // (self._clock_in_avg * 24))))
if new_bpm != self.bpm: self.bpm = new_bpm
if new_bpm != self.bpm:
self.bpm = new_bpm; self._beat_ns = 60_000_000_000 // new_bpm; self._rebuild_dur_all()
self._slaved = True
def _slave_start(self): # master sent Start (or Continue) -> start playback
if not self.running:
@ -1457,20 +1501,19 @@ class App:
try: os.remove("/trial")
except Exception: pass
committed = True
# Refresh display ~30x/s, BUT skip if a beat is due within ~35ms (refresh() blocks SPI for that long)
# so the click never gets pushed late by the redraw. Force-refresh after 0.5s anyway so visuals stay live.
# Refresh display ~20x/s, skip ONLY when the MASTER lane's next step is within ~10ms (its alignment
# matters most musically; sub-lanes can take a ~few ms jitter without audible problem). Force-refresh
# after 200ms so titles + meters still feel live at fast subdivisions.
if self.dirty and tnow >= self._refreshNext:
safe = True
if self.running and self.lanes:
now_ns = time.monotonic_ns(); next_beat = self.lanes[0]['next']
for L in self.lanes:
if L['next'] < next_beat: next_beat = L['next']
safe = (next_beat - now_ns) > 35_000_000 or (tnow - self._lastRefresh) > 0.5
nb = self.lanes[0]['next'] # master only -> doesn't starve at fine subdivisions
safe = (nb - time.monotonic_ns()) > 10_000_000 or (tnow - self._lastRefresh) > 0.2
if safe:
if self.display.refresh(): self.dirty = False
self._lastRefresh = tnow; self._refreshNext = tnow + 0.033
self._lastRefresh = tnow; self._refreshNext = tnow + 0.05
else:
self._refreshNext = tnow + 0.005 # check again soon; don't wait the full 33ms
self._refreshNext = tnow + 0.003 # check again very soon; don't wait the 50ms
time.sleep(0.0005)
except MemoryError: # surface, gc, keep running (don't crash on a fragmented heap)
try: print("MemoryError: gc + continue")