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

Firmware push fix on both Kit (0.0.24) + Explorer (0.0.4)

Browse source 
Diagnosed from the user's console output - 25 chunks pushed cleanly at
~124ms each, then stalled. Two coupled causes:

1) Bus contention. tick() and Live sync share self.midi with the chunk
   ACKs. While the device was processing a chunk, a Note On / Clock Out /
   Live-sync FULL heartbeat could land on the same MIDI OUT stream and
   the host's parser dropped the interleaved ACK SysEx.

   Fix: self._fw_pushing flag set on 0x21 BEGIN, cleared on 0x23 COMMIT
   or any error. midi_send / Clock Out / _sync_broadcast / _sync_broadcast_full
   all early-out when _fw_pushing is True. Only ACKs go out during a push.

2) SysEx assembler garbage. self._sx = bytearray() per chunk leaks ~70
   bytes / chunk that only GC'd every 50 chunks. 25 chunks of trash plus a
   slow heap walked the wrong way explains the ramp-up to 174 -> 119 -> 124
   ms ACK times. GC every chunk now (~30ms cost on RP2040/RP2350 with
   small heap) so the assembler buffer is always fresh.

Same patch on both pico-cp/ and pico-explorer/ since the bug is identical.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Me Here 2026-05-30 22:51:46 -05:00
parent 3805c5ee00
commit c625a8aaa2
6 changed files with 158 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

66
hardware/eda/circuits/rtc.py Normal file
View file

@ -0,0 +1,66 @@
#!/usr/bin/env python3
"""PM_K-1 RTC (SKiDL): RV-8803-C7 I2C real-time clock for the practice-log timestamps.
Run INSIDE the EDA container:
cd hardware/eda && ./run.sh python3 ../eda/circuits/rtc.py
Outputs ERC + hardware/kicad/rtc.net.
VERIFIED pinout (Micro Crystal RV-8803-C7, 8-WCDFN 3.2x1.5mm):
1=SDA 2=CLKOUT 3=VDD 4=CLKOE 5=VSS 6=/INT 7=EVI 8=SCL. Single VDD (no separate VBACKUP
pin); ~240nA typ. Shares the touch I2C bus (SDA=GPIO8, SCL=GPIO9).
BACKUP: diode-OR -- system +3V3 OR the CR2032 feed VDD_RTC through Schottkys, so the
board runs the RTC off 3V3 when on, the coin cell only when off, and (importantly for an
heirloom) the cell can be replaced WITHOUT the RTC losing time. Schottkys block charging
the (non-rechargeable) cell.
Unused pins: CLKOE->GND (CLKOUT disabled), CLKOUT->NC, EVI->GND (no event input),
/INT pulled up + routed to RTC_INT (optional alarm IRQ to a GPIO).
CONFIRM at layout: the RV-8803-C7 footprint, and cross-check the App Manual's recommended
backup circuit.
"""
import os
from skidl import *
set_default_tool(KICAD9)
P = Pin.types
R = Part("Device","R", dest=TEMPLATE, footprint="Resistor_SMD:R_0402_1005Metric")
def C(v): return Part("Device","C", value=v, footprint="Capacitor_SMD:C_0402_1005Metric")
DS = Part("Device","D_Schottky", dest=TEMPLATE, footprint="Diode_SMD:D_SOD-323")
p3v3, gnd = Net("+3V3"), Net("GND")
p3v3.drive = POWER; gnd.drive = POWER
i2c_sda, i2c_scl = Net("I2C_SDA"), Net("I2C_SCL") # shared bus (RP2350 GPIO8/9 + touch)
vdd_rtc, rtc_int = Net("VDD_RTC"), Net("RTC_INT")
RV8803 = Part(name="RV-8803-C7", tool=SKIDL, dest=TEMPLATE, ref_prefix="U",
footprint="RTC_MicroCrystal:RV-8803-C7", # footprint: confirm at layout
pins=[Pin(num=1,name="SDA",func=P.BIDIR),Pin(num=2,name="CLKOUT",func=P.OUTPUT),
Pin(num=3,name="VDD",func=P.PWRIN),Pin(num=4,name="CLKOE",func=P.INPUT),
Pin(num=5,name="VSS",func=P.PWRIN),Pin(num=6,name="INT",func=P.OPENCOLL),
Pin(num=7,name="EVI",func=P.INPUT),Pin(num=8,name="SCL",func=P.INPUT)])
u = RV8803(ref="U7")
u["VDD"] += vdd_rtc; u["VSS"] += gnd
u["SDA"] += i2c_sda; u["SCL"] += i2c_scl
u["CLKOE"] += gnd # disable CLKOUT
u["EVI"] += gnd # unused event input
u["INT"] += rtc_int # open-drain alarm (optional)
# CLKOUT left unconnected
# diode-OR backup: 3V3 -> D1 -> VDD_RTC ; CR2032 -> D2 -> VDD_RTC
d1, d2 = DS(value="BAT54"), DS(value="BAT54")
p3v3 += d1[2]; d1[1] += vdd_rtc # D_Schottky pin1=K, pin2=A : anode at 3V3, cathode at VDD_RTC
bt = Part("Device","Battery_Cell", value="CR2032",
footprint="Battery:BatteryHolder_Keystone_1066_1x2032", ref="BT1")
bt["+"] += d2[2]; d2[1] += vdd_rtc; bt["-"] += gnd
crtc = C("100nF"); vdd_rtc += crtc[1]; crtc[2] += gnd
# I2C pull-ups (bus shared with touch) + /INT pull-up, to 3V3
for net in (i2c_sda, i2c_scl):
r = R(value="4.7k"); net += r[1]; r[2] += p3v3
rint = R(value="10k"); rtc_int += rint[1]; rint[2] += p3v3
ERC()
out = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "kicad", "rtc.net"))
generate_netlist(file_=out)
print("RTC netlist ->", out)

5
hardware/kicad/rtc.erc Normal file
View file

@ -0,0 +1,5 @@
ERC WARNING: Insufficient drive current on net VDD_RTC for pin POWER-IN pin 3/VDD of RV-8803-C7/U7.
ERC WARNING: Unconnected pin: OUTPUT pin 2/CLKOUT of RV-8803-C7/U7.
ERC INFO: 2 warnings found while running ERC.
ERC INFO: 0 errors found while running ERC.

27
hardware/kicad/rtc.log Normal file
View file

@ -0,0 +1,27 @@
WARNING: KICAD8_SYMBOL_DIR environment variable is missing, so the default KiCad symbol libraries won't be searched. @ [/work/hardware/kicad/<frozen importlib._bootstrap_external>:995=>/work/hardware/kicad/<frozen importlib._bootstrap>:488]
WARNING: KICAD6_SYMBOL_DIR environment variable is missing, so the default KiCad symbol libraries won't be searched. @ [/work/hardware/kicad/<frozen importlib._bootstrap_external>:995=>/work/hardware/kicad/<frozen importlib._bootstrap>:488]
WARNING: KICAD7_SYMBOL_DIR environment variable is missing, so the default KiCad symbol libraries won't be searched. @ [/work/hardware/kicad/<frozen importlib._bootstrap_external>:995=>/work/hardware/kicad/<frozen importlib._bootstrap>:488]
WARNING: fp-lib-table file was not found. Component footprints are not available.
WARNING: fp-lib-table file was not found. Component footprints are not available.
WARNING: fp-lib-table file was not found. Component footprints are not available.
WARNING: fp-lib-table file was not found. Component footprints are not available.
WARNING: Missing tag on RV-8803-C7 instantiated at /work/hardware/eda/circuits/rtc.py:41.
WARNING: Random tag vvCZlacOIl generated for RV-8803-C7.
WARNING: Missing tag on D_Schottky instantiated at /work/hardware/eda/circuits/rtc.py:51.
WARNING: Random tag _zzfn3XmHQ generated for D_Schottky.
WARNING: Missing tag on D_Schottky instantiated at /work/hardware/eda/circuits/rtc.py:51.
WARNING: Random tag c6BuDsK_A2 generated for D_Schottky.
WARNING: Missing tag on Battery_Cell instantiated at /work/hardware/eda/circuits/rtc.py:53.
WARNING: Random tag n7hVbpl_7H generated for Battery_Cell.
WARNING: Missing tag on C instantiated at /work/hardware/eda/circuits/rtc.py:27.
WARNING: Random tag 0kNu6jvDLR generated for C.
WARNING: Missing tag on R instantiated at /work/hardware/eda/circuits/rtc.py:60.
WARNING: Random tag _uIfBEh0dY generated for R.
WARNING: Missing tag on R instantiated at /work/hardware/eda/circuits/rtc.py:60.
WARNING: Random tag 4HsfrE7DU9 generated for R.
WARNING: Missing tag on R instantiated at /work/hardware/eda/circuits/rtc.py:61.
WARNING: Random tag S0w81WPsYC generated for R.
WARNING: Missing tag on instantiated at /work/hardware/kicad/<frozen importlib._bootstrap>:488.
INFO: 21 warnings found while generating netlist.
INFO: 0 errors found while generating netlist.

29
hardware/kicad/rtc_sklib.py Normal file
View file

@ -0,0 +1,29 @@
from collections import defaultdict
from skidl import Pin, Part, Alias, SchLib, SKIDL, TEMPLATE
from skidl.pin import pin_types
SKIDL_lib_version = '0.0.1'
rtc = SchLib(tool=SKIDL).add_parts(*[
Part(**{ 'name':'RV-8803-C7', 'dest':TEMPLATE, 'tool':SKIDL, 'aliases':Alias({'RV-8803-C7'}), 'ref_prefix':'U', 'fplist':None, 'footprint':'RTC_MicroCrystal:RV-8803-C7', 'keywords':None, 'description':'', 'datasheet':None, 'pins':[
Pin(num='1',name='SDA',func=pin_types.BIDIR),
Pin(num='2',name='CLKOUT',func=pin_types.OUTPUT),
Pin(num='3',name='VDD',func=pin_types.PWRIN),
Pin(num='4',name='CLKOE',func=pin_types.INPUT),
Pin(num='5',name='VSS',func=pin_types.PWRIN),
Pin(num='6',name='INT',func=pin_types.OPENCOLL),
Pin(num='7',name='EVI',func=pin_types.INPUT),
Pin(num='8',name='SCL',func=pin_types.INPUT)] }),
Part(**{ 'name':'D_Schottky', 'dest':TEMPLATE, 'tool':SKIDL, 'aliases':Alias({'D_Schottky'}), 'ref_prefix':'D', 'fplist':[''], 'footprint':'Diode_SMD:D_SOD-323', 'keywords':'diode Schottky', 'description':'Schottky diode', 'datasheet':'~', 'pins':[
Pin(num='1',name='K',func=pin_types.PASSIVE,unit=1),
Pin(num='2',name='A',func=pin_types.PASSIVE,unit=1)], 'unit_defs':[] }),
Part(**{ 'name':'Battery_Cell', 'dest':TEMPLATE, 'tool':SKIDL, 'aliases':Alias({'Battery_Cell'}), 'ref_prefix':'BT', 'fplist':[''], 'footprint':'Battery:BatteryHolder_Keystone_1066_1x2032', 'keywords':'battery cell', 'description':'Single-cell battery', 'datasheet':'~', 'pins':[
Pin(num='1',name='+',func=pin_types.PASSIVE,unit=1),
Pin(num='2',name='-',func=pin_types.PASSIVE,unit=1)], 'unit_defs':[] }),
Part(**{ 'name':'C', 'dest':TEMPLATE, 'tool':SKIDL, 'aliases':Alias({'C'}), 'ref_prefix':'C', 'fplist':[''], 'footprint':'Capacitor_SMD:C_0402_1005Metric', 'keywords':'cap capacitor', 'description':'Unpolarized capacitor', 'datasheet':'~', 'pins':[
Pin(num='1',name='~',func=pin_types.PASSIVE,unit=1),
Pin(num='2',name='~',func=pin_types.PASSIVE,unit=1)], 'unit_defs':[] }),
Part(**{ 'name':'R', 'dest':TEMPLATE, 'tool':SKIDL, 'aliases':Alias({'R'}), 'ref_prefix':'R', 'fplist':[''], 'footprint':'Resistor_SMD:R_0402_1005Metric', 'keywords':'R res resistor', 'description':'Resistor', 'datasheet':'~', 'pins':[
Pin(num='1',name='~',func=pin_types.PASSIVE,unit=1),
Pin(num='2',name='~',func=pin_types.PASSIVE,unit=1)], 'unit_defs':[] })])

29
pico-cp/app.py
View file

@ -18,7 +18,7 @@
import board, busio, digitalio, analogio, pwmio, displayio, vectorio, time, json, gc, os, supervisor 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 supervisor.runtime.autoreload = False # we write our own files (log + pushed programs); never self-restart
APP_VERSION = "0.0.23" # firmware version (the A/B updater pushes/compares this) APP_VERSION = "0.0.24" # firmware version (the A/B updater pushes/compares this)
DEVICE_ID = "K" # 'K' = 52Pi kit, 'X' = Pimoroni Explorer (per docs/livesync-protocol.md and the version reply) DEVICE_ID = "K" # 'K' = 52Pi kit, 'X' = Pimoroni Explorer (per docs/livesync-protocol.md and the version reply)
try: try:
import rtc # set from the editor's clock SysEx so the log has real timestamps import rtc # set from the editor's clock SysEx so the log has real timestamps
@ -439,7 +439,7 @@ class App:
self.midi_in = usb_midi.ports[0] if (MIDI_ENABLED and usb_midi and len(usb_midi.ports) > 0) 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._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._sx = bytearray(); self._sxon = False # USB-MIDI SysEx assembler (clock + pushed programs)
self._fw = None; self._fw_n = 0 # chunked firmware transfer: staging file handle + chunk counter self._fw = None; self._fw_n = 0; self._fw_pushing = False # chunked firmware transfer state + bus-quiet flag
self.led = RGB(P_RGB) self.led = RGB(P_RGB)
self.spk = pwmio.PWMOut(P_SPK, frequency=1600, variable_frequency=True, duty_cycle=0) self.spk = pwmio.PWMOut(P_SPK, frequency=1600, variable_frequency=True, duty_cycle=0)
self.spk_off = 0 self.spk_off = 0
@ -1006,11 +1006,11 @@ class App:
try: self.midi.write(b) try: self.midi.write(b)
except Exception: pass except Exception: pass
def _sync_broadcast(self, evt): # one DELTA event; suppressed while applying a remote change (echo guard) def _sync_broadcast(self, evt): # one DELTA event; suppressed while applying a remote change (echo guard)
if not self._sync_armed or self._sync_applying or self.midi is None: return if not self._sync_armed or self._sync_applying or self.midi is None or self._fw_pushing: return
text = "%s;%d;%s" % (self._sync_origin, self._sync_seq, evt); self._sync_seq += 1 text = "%s;%d;%s" % (self._sync_origin, self._sync_seq, evt); self._sync_seq += 1
self._sync_send(0x42, text) self._sync_send(0x42, text)
def _sync_broadcast_full(self): # FULL snapshot: running + sl + item + patch (coalesces structural edits) def _sync_broadcast_full(self): # FULL snapshot: running + sl + item + patch (coalesces structural edits)
if not self._sync_armed or self.midi is None: return if not self._sync_armed or self.midi is None or self._fw_pushing: return
try: patch = self._prog_str() try: patch = self._prog_str()
except Exception: return except Exception: return
text = "%s;%d;%d;%d;%d;%s" % (self._sync_origin, self._sync_seq, text = "%s;%d;%d;%d;%d;%s" % (self._sync_origin, self._sync_seq,
@ -1110,7 +1110,7 @@ class App:
self._sync_applying = False self._sync_applying = False
def midi_send(self, note, vel): # device-as-conductor: a note per click to the computer def midi_send(self, note, vel): # device-as-conductor: a note per click to the computer
if self.midi is None: return if self.midi is None or self._fw_pushing: return # keep the bus quiet during a firmware push so ACKs aren't interleaved
b = self._note_buf # reused bytearray -> zero alloc per click (hot path) b = self._note_buf # reused bytearray -> zero alloc per click (hot path)
b[0] = 0x90 | ((MIDI_CHANNEL - 1) & 0x0F) # Note On, channel 1..16 b[0] = 0x90 | ((MIDI_CHANNEL - 1) & 0x0F) # Note On, channel 1..16
b[1] = note & 0x7F; b[2] = vel & 0x7F b[1] = note & 0x7F; b[2] = vel & 0x7F
@ -1205,7 +1205,7 @@ class App:
self._advance = False self._advance = False
self._do_advance() self._do_advance()
# MIDI Clock Out (master): 24 PPQN; interval follows the live bpm (so continuous ramps carry through) # MIDI Clock Out (master): 24 PPQN; interval follows the live bpm (so continuous ramps carry through)
if self.running and MIDI_CLOCK_OUT and self.midi is not None and not self._slaved: # don't echo to the master if self.running and MIDI_CLOCK_OUT and self.midi is not None and not self._slaved and not self._fw_pushing:
clk = self._clock_byte # reused singleton bytes (no per-tick alloc) clk = self._clock_byte # reused singleton bytes (no per-tick alloc)
tick_ns = self._beat_ns // 24 # cached: ns per Clock pulse tick_ns = self._beat_ns // 24 # cached: ns per Clock pulse
while now >= self._clock_next: while now >= self._clock_next:
@ -1579,21 +1579,23 @@ class App:
try: try:
try: self._fw.close() try: self._fw.close()
except Exception: pass except Exception: pass
self._fw = open("/app.new", "wb"); self._fw_n = 0; self._ack(True) self._fw = open("/app.new", "wb"); self._fw_n = 0
self._fw_pushing = True # silence Note On / Clock Out / Live-sync broadcasts during the push
self._ack(True)
except Exception: # read-only (editor mode) / no space except Exception: # read-only (editor mode) / no space
self._fw = None; self._ack(False) self._fw = None; self._fw_pushing = False; self._ack(False)
elif cmd == 0x22: # DATA: a base64 chunk (multiple of 4) -> decode -> append elif cmd == 0x22: # DATA: a base64 chunk (multiple of 4) -> decode -> append
try: try:
if self._fw is None or a2b_base64 is None: raise OSError() if self._fw is None or a2b_base64 is None: raise OSError()
self._fw.write(a2b_base64(bytes(sx[2:]))) self._fw.write(a2b_base64(bytes(sx[2:])))
self._fw.flush() # small, predictable per-chunk flush (no slow burst flushes later) self._fw.flush() # small, predictable per-chunk flush (no slow burst flushes later)
self._fw_n += 1 self._fw_n += 1
if self._fw_n % 50 == 0: gc.collect() # keep the heap fresh during a long push gc.collect() # SysEx assembler allocates a fresh bytearray per chunk;
self._ack(True) self._ack(True) # GC every chunk so 600 chunks' worth of garbage doesn't accumulate
except Exception: except Exception:
try: self._fw.close() try: self._fw.close()
except Exception: pass except Exception: pass
self._fw = None; self._ack(False) self._fw = None; self._fw_pushing = False; self._ack(False)
elif cmd == 0x23: # COMMIT: verify it's a CircuitPython .mpy, then A/B install elif cmd == 0x23: # COMMIT: verify it's a CircuitPython .mpy, then A/B install
try: try:
try: self._fw.close() try: self._fw.close()
@ -1603,15 +1605,16 @@ class App:
if os.stat("/app.new")[6] < 4000 or len(head) < 2 or head[0] != 0x43 or head[1] != 0x06: 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 try: os.remove("/app.new") # not a CircuitPython mpy v6 -> reject, keep the working build
except OSError: pass except OSError: pass
self._ack(False); return self._fw_pushing = False; self._ack(False); return
try: os.remove("/app.bak") try: os.remove("/app.bak")
except OSError: pass except OSError: pass
os.rename("/app.mpy", "/app.bak") # current build becomes the rollback os.rename("/app.mpy", "/app.bak") # current build becomes the rollback
os.rename("/app.new", "/app.mpy") os.rename("/app.new", "/app.mpy")
open("/trial", "w").close() # arm the trial; the loader reverts if it won't boot open("/trial", "w").close() # arm the trial; the loader reverts if it won't boot
self._fw_pushing = False
self._ack(True); time.sleep(0.4); supervisor.reload() self._ack(True); time.sleep(0.4); supervisor.reload()
except Exception: # catch ALL (read-only, MemoryError, ...) -> never brick except Exception: # catch ALL (read-only, MemoryError, ...) -> never brick
self._ack(False) self._fw_pushing = False; self._ack(False)
def _ack(self, ok): def _ack(self, ok):
if self.midi: self.midi.write(bytes([0xF0, 0x7D, 0x7F if ok else 0x7E, 0xF7])) if self.midi: self.midi.write(bytes([0xF0, 0x7D, 0x7F if ok else 0x7E, 0xF7]))

27
pico-explorer/app.py
View file

@ -339,7 +339,7 @@ class App:
self.midi_in = usb_midi.ports[0] if (MIDI_ENABLED and usb_midi and len(usb_midi.ports) > 0) 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._mbuf = bytearray(64); self.midi_host = False; self.last_midi_in = 0.0
self._sx = bytearray(); self._sxon = False # USB-MIDI SysEx assembler self._sx = bytearray(); self._sxon = False # USB-MIDI SysEx assembler
self._fw = None; self._fw_n = 0 # chunked firmware transfer state self._fw = None; self._fw_n = 0; self._fw_pushing = False # chunked firmware transfer state + bus-quiet flag
self.spk = pwmio.PWMOut(P_AUDIO, frequency=1600, variable_frequency=True, duty_cycle=0) self.spk = pwmio.PWMOut(P_AUDIO, frequency=1600, variable_frequency=True, duty_cycle=0)
self.amp_en = digitalio.DigitalInOut(P_AMPEN); self.amp_en.direction = digitalio.Direction.OUTPUT self.amp_en = digitalio.DigitalInOut(P_AMPEN); self.amp_en.direction = digitalio.Direction.OUTPUT
self._amp(False) # amp off when no audio playing (saves power, kills hum) self._amp(False) # amp off when no audio playing (saves power, kills hum)
@ -737,11 +737,11 @@ class App:
try: self.midi.write(b) try: self.midi.write(b)
except Exception: pass except Exception: pass
def _sync_broadcast(self, evt): def _sync_broadcast(self, evt):
if not self._sync_armed or self._sync_applying or self.midi is None: return if not self._sync_armed or self._sync_applying or self.midi is None or self._fw_pushing: return
text = "%s;%d;%s" % (self._sync_origin, self._sync_seq, evt); self._sync_seq += 1 text = "%s;%d;%s" % (self._sync_origin, self._sync_seq, evt); self._sync_seq += 1
self._sync_send(0x42, text) self._sync_send(0x42, text)
def _sync_broadcast_full(self): def _sync_broadcast_full(self):
if not self._sync_armed or self.midi is None: return if not self._sync_armed or self.midi is None or self._fw_pushing: return
try: patch = self._prog_str() try: patch = self._prog_str()
except Exception: return except Exception: return
text = "%s;%d;%d;%d;%d;%s" % (self._sync_origin, self._sync_seq, text = "%s;%d;%d;%d;%d;%s" % (self._sync_origin, self._sync_seq,
@ -845,7 +845,7 @@ class App:
L['levels'] = [(2 if (i // sub) in starts else 1) if i % sub == 0 else 0 for i in range(L['steps'])] L['levels'] = [(2 if (i // sub) in starts else 1) if i % sub == 0 else 0 for i in range(L['steps'])]
def midi_send(self, note, vel): def midi_send(self, note, vel):
if self.midi is None: return if self.midi is None or self._fw_pushing: return # keep the bus quiet during a firmware push so ACKs aren't interleaved
b = self._note_buf b = self._note_buf
b[0] = 0x90 | ((MIDI_CHANNEL - 1) & 0x0F) b[0] = 0x90 | ((MIDI_CHANNEL - 1) & 0x0F)
b[1] = note & 0x7F; b[2] = vel & 0x7F b[1] = note & 0x7F; b[2] = vel & 0x7F
@ -935,7 +935,7 @@ class App:
if self._advance: if self._advance:
self._advance = False self._advance = False
self._do_advance() self._do_advance()
if self.running and MIDI_CLOCK_OUT and self.midi is not None and not self._slaved: if self.running and MIDI_CLOCK_OUT and self.midi is not None and not self._slaved and not self._fw_pushing:
clk = self._clock_byte clk = self._clock_byte
tick_ns = self._beat_ns // 24 tick_ns = self._beat_ns // 24
while now >= self._clock_next: while now >= self._clock_next:
@ -1391,21 +1391,23 @@ class App:
try: try:
try: self._fw.close() try: self._fw.close()
except Exception: pass except Exception: pass
self._fw = open("/app.new", "wb"); self._fw_n = 0; self._ack(True) self._fw = open("/app.new", "wb"); self._fw_n = 0
self._fw_pushing = True # silence Note On / Clock Out / Live-sync broadcasts during the push
self._ack(True)
except Exception: except Exception:
self._fw = None; self._ack(False) self._fw = None; self._fw_pushing = False; self._ack(False)
elif cmd == 0x22: elif cmd == 0x22:
try: try:
if self._fw is None or a2b_base64 is None: raise OSError() if self._fw is None or a2b_base64 is None: raise OSError()
self._fw.write(a2b_base64(bytes(sx[2:]))) self._fw.write(a2b_base64(bytes(sx[2:])))
self._fw.flush() self._fw.flush()
self._fw_n += 1 self._fw_n += 1
if self._fw_n % 50 == 0: gc.collect() gc.collect() # the SysEx assembler allocates a fresh bytearray per chunk -
self._ack(True) self._ack(True) # GC every chunk so 600 chunks' worth of garbage doesn't accumulate
except Exception: except Exception:
try: self._fw.close() try: self._fw.close()
except Exception: pass except Exception: pass
self._fw = None; self._ack(False) self._fw = None; self._fw_pushing = False; self._ack(False)
elif cmd == 0x23: elif cmd == 0x23:
try: try:
try: self._fw.close() try: self._fw.close()
@ -1415,15 +1417,16 @@ class App:
if os.stat("/app.new")[6] < 4000 or len(head) < 2 or head[0] != 0x43 or head[1] != 0x06: if os.stat("/app.new")[6] < 4000 or len(head) < 2 or head[0] != 0x43 or head[1] != 0x06:
try: os.remove("/app.new") try: os.remove("/app.new")
except OSError: pass except OSError: pass
self._ack(False); return self._fw_pushing = False; self._ack(False); return
try: os.remove("/app.bak") try: os.remove("/app.bak")
except OSError: pass except OSError: pass
os.rename("/app.mpy", "/app.bak") os.rename("/app.mpy", "/app.bak")
os.rename("/app.new", "/app.mpy") os.rename("/app.new", "/app.mpy")
open("/trial", "w").close() open("/trial", "w").close()
self._fw_pushing = False
self._ack(True); time.sleep(0.4); supervisor.reload() self._ack(True); time.sleep(0.4); supervisor.reload()
except Exception: except Exception:
self._ack(False) self._fw_pushing = False; self._ack(False)
def _ack(self, ok): def _ack(self, ok):
if self.midi: self.midi.write(bytes([0xF0, 0x7D, 0x7F if ok else 0x7E, 0xF7])) if self.midi: self.midi.write(bytes([0xF0, 0x7D, 0x7F if ok else 0x7E, 0xF7]))