From 1568eac099cff89942ff04995fce1ddb079ebe0c Mon Sep 17 00:00:00 2001 From: John Remillard Date: Tue, 21 Jul 2026 23:05:26 -0400 Subject: [PATCH 1/4] feat: implement high-precision WebSocket transport BPM updates and bypass MIDI CC emission for bound transport parameters --- modalapi/modhandler.py | 18 +- modalapi/pedalboard.py | 3 +- .../bpm_dialog_121.png | Bin 0 -> 8476 bytes tests/v3/test_transport_bindings.py | 249 ++++++++++++++++++ 4 files changed, 266 insertions(+), 4 deletions(-) create mode 100644 tests/snapshots/v3/test_transport_bindings/test_encoder_bpm_turn_parameter_dialog_snapshot/bpm_dialog_121.png diff --git a/modalapi/modhandler.py b/modalapi/modhandler.py index f5d9f201b..9cddd3f76 100755 --- a/modalapi/modhandler.py +++ b/modalapi/modhandler.py @@ -419,6 +419,8 @@ def _handle_encoder(self, event: EncoderEvent) -> bool: new_value = ParameterSteps.for_parameter(c.parameter).move(delta) c.parameter.value = new_value self.lcd.display_parameter_value(c.parameter, new_value) + if c.parameter.instance_id == Pedalboard.TRANSPORT_INSTANCE_ID and c.parameter.symbol == BPM_SYMBOL: + self.set_mod_tap_tempo(new_value) emit_value = c.bar_midi_value() else: emit_value = self._advance_encoder_fallback(c, delta) @@ -426,7 +428,11 @@ def _handle_encoder(self, event: EncoderEvent) -> bool: # Unconditional, and must stay that way: an unbound encoder has no row, # and this emit is the only way mod-ui sees its CC to MIDI-learn it. # Emission is hardware-level, below the table (see input/README.md). - self._emit_midi(c, emit_value) + # Transport BPM parameters bypass 7-bit MIDI CC emission for high-precision WebSocket transport. + if c.parameter is None or not ( + c.parameter.instance_id == Pedalboard.TRANSPORT_INSTANCE_ID and c.parameter.symbol == BPM_SYMBOL + ): + self._emit_midi(c, emit_value) return True def encoder_fallback(self, controller: EncoderController) -> int: @@ -1427,6 +1433,9 @@ def parameter_value_commit(self, param, value): # Audio parameter (volume, EQ, etc.) - handled locally, no remote update needed if param.instance_id is None: self.audio_parameter_commit(param.symbol, value) + # Pedalboard-level transport BPM parameters are set via WebSocket for precision + if param.instance_id == Pedalboard.TRANSPORT_INSTANCE_ID and param.symbol == BPM_SYMBOL: + self.set_mod_tap_tempo(value) return # External MIDI parameters have no mod-host counterpart. The dialog's NAV @@ -1442,7 +1451,7 @@ def parameter_value_commit(self, param, value): self._emit_midi(controller, int(value)) return - if not self._is_pedalboard_loading: + if not self._is_pedalboard_loading and param.instance_id is not None: self.ws_bridge.send_parameter(param.instance_id, param.symbol, param.value) @property @@ -1798,8 +1807,11 @@ def audio_parameter_commit(self, symbol, value): def get_callback(self, callback_name): return util.DICT_GET(self.callbacks, callback_name) - def set_mod_tap_tempo(self, bpm): + def set_mod_tap_tempo(self, bpm: float | None) -> None: if bpm is not None: + self._last_bpm_change_time = time.time() + if self.ws_bridge is not None: + self.ws_bridge.send_bpm(bpm) self._rest_post(self.root_uri + "set_bpm", json={"value": bpm}) def set_sync_mode(self, mode: SyncMode) -> None: diff --git a/modalapi/pedalboard.py b/modalapi/pedalboard.py index 7611a82d6..b9cd080cf 100755 --- a/modalapi/pedalboard.py +++ b/modalapi/pedalboard.py @@ -21,7 +21,7 @@ from typing import Optional -from common.parameter import BYPASS_SYMBOL, Parameter, PortInfo, Symbol, json_default +from common.parameter import BYPASS_SYMBOL, TTL_INTEGER, Parameter, PortInfo, Symbol, json_default import modalapi.plugin as Plugin from modalapi.connections import Connection, build_connection from modalapi.plugin_customization import Customizer, default_customizer @@ -57,6 +57,7 @@ def _transport_port_info(symbol: Symbol) -> PortInfo: symbol=":bpm", ranges={"minimum": _BPM_RANGE[0], "maximum": _BPM_RANGE[1]}, units={"symbol": "BPM", "label": "beats per minute"}, + properties=[TTL_INTEGER], ) if symbol == BPB_SYMBOL: return PortInfo( diff --git a/tests/snapshots/v3/test_transport_bindings/test_encoder_bpm_turn_parameter_dialog_snapshot/bpm_dialog_121.png b/tests/snapshots/v3/test_transport_bindings/test_encoder_bpm_turn_parameter_dialog_snapshot/bpm_dialog_121.png new file mode 100644 index 0000000000000000000000000000000000000000..665e88e3ef08422f8d0cfa746774480c86cc4074 GIT binary patch literal 8476 zcmai)by!qU_wQ%ulu`twTN`lIk;F(_fpoQb<@G^{qv0V5Nm-oO#wWHBxAS72{*f2nj|t`a8D^=E zs&K^mXB#@8a5@=Nmp(r%EG#tHprD)X@%#cy7;I%_rMs(3BU?NtJ32l- zK0Q6%-zUZkM+=2O*x!vg)NNkD=bkaf(`?kcEp_$uJcID5y5{mB`r`uw1JTgX6l1BZ zg-z*+qn!5Sm2|{J*lT~hR zZWI`Q6}|Yv$i%d@xtWofNpqX2oIh4sS(%bTUL&yex*7DivXT!%VHhDVFMkzae{TjR zynqYF1AG7EWbNp}zpWj-ii?Xo2H4*@2*(Ys;f22$*|@sA6qiJ}xVSJdFqD^<3keCO zR7{SK@B6mK#SzyT{&Z3&Pzh{Aq+Oz>%ryS~rLMaP`y?uN>#>A8r71=pD)?5{i&}eVfK4*5Yx96wvJli#?u@{$+ z$k#@FvXQXiX3xyZ^ zs{az48y_E^oK&Q&+?aON*H4|n>R_odaB6yXlP+NW?v+0hiR3kDwi*mLFt%GYA5zvI zs%&WRsfBvtVSAGI5syif>2M_LaO}7mrOt$=K4(@t`U3|>I5e-7O>KRo7PEx;^CO~6 z-n@OQfCZe5(EAy&V$j*yd83ZFKmTK?>UPt#%{1ir@2}k8%YoA_bHbQWt&GuhPqCp* zi%?l&T3Vj?=Ov}3&NFVlr!%%^JAWEHH#A`WyvB`uMO};1e3tPU8B_(An@7~Rr&`52 zps#s(dBD*e$?xo8o<2UdhB=&^oX)es!lpP+k;exseb||%1bmjMAaWBDNEgwPRdltr z8B+(_+S*u1acf%UM@L6zXPvGzu0Y$w#6;-g{!(}S-dP4?-vY556G6bLEDl2$vE2WD z7=I`k-_3uzw;*FPL}7D&xHgz6?ChcSGz%G=df`te`tb6n$bxaDdFTOZFXZluAuB67 zD`Mu-e0*+hZhpSqX{I`!E5?8;=ZWp`c=5|GpFe};K~hQz%xoh=LkH*ZZ<*jiif z%+%P5uDjk1;sXDh85y;J$m!|n`Sgiijn~8a>C>lRinaLceJd|F=?Io8bzB7PE6h4_ zW~w>3xQ2#@+XF8=OiXD0`0L2pS>a&3`Sj^iA8mlse5)T=oWK`zqykRC-{IlmG=(|L z|K-bz!sYrGqZPCzh6$t!-@*5)z;v zJ+*W__$si>+dHDi(Eg&Wm7 z@y{y~LTrZ8I>+01(&j@nBO2KS_Ki6Nr$@=U=~Cq7xF~(yiN*ypW!8RFhYd0qVBIwe zHC`O~?*Fw*lfksn#Hd>8wV}JC2|myFK{Mk`lfcc%g2dwy?=R($@+8S!irKj|c-<@W zI#~V=CP?ffk>ND{FGWSqSrfna^~oIckY$Rvg26eA61G&&5F8L}Gb|UTr~g!0N!*ZS z5;=}8cY}5FjUVrPxq&%Y?Ubvn3FFq8Rt*n(3 z1UNW+-aU^=qv!p%xYwm!nSSGyYZu?{N6HN4%7mP3PJm?;5|^DVwc5n+@F{X%RV*+A z8aZ+*^%VB#n&qMC1>SlHj^i!aCP9Pj{>dBLfF zoHr5NyWverpIx_Ef>kc(BFxb0xZQTK(%%;ACoW#F`A2x&uPqwD^Wh^Z!QQkvuELpa zEZSpKz?MJc(&kl`lFbiwOpfy=S8;H@dA6e6(Mm+F7G7BRrur=bQ6tx{&V9N&Ijr^; zg4~%eCWogc)B6tv;v`G=aJ{qt((n|ASKL9ja#_2d3K(r5Jk-?WMp&*oV!0nflv1hvUKCAX_aF&5yD6% zOdH17-Bw4g{#InSLxd47S>{F-=ftcLfr33Z*U8Ac|Ck5QTT;`;nrkePY%!XB+B;@7 zP%HOb1#QNQ*{5QM;<(!2;X{Sf+$+z#`}5+GNi!-vQs{L=v$0X#n*{s^XJ;aWl~qhv zBzSJl&gH0`>6nHvj9vDU2%ipC^gU^8GI`~42_Mo`W0os(IjD|MXG>pRb$ra-Hp{*T zn}p5XsqI+Vp$eMcL-p>-F*Tzef9E(P@^+}v zrEj+v#@q`nRvN*ZW25~j7@#8MpOf)Tldxny*>n}FCFD;_x6O}6LdjajblP$L;N~Gu zw_JnyQ3=V#YEiA%d^7v;QF%3w>S@DsT_l=ioz|M6r>V!as{Fb3+>Tdw=}bu3D9gJbomb zTBBF6{;B=VsEj+p+QOl0)@NYlCyNveBhWXHnkz^6xX9?t03)zD;j4Z=17T#E>6UQ= z)(70TD|Z!VsGSxoqQtSpZVI);y

P+P|xmE*~{eO9)DVoIa*L8?L#sw=F7K&y^id zeKL1mejVr1W`^e0no#nuJpP|2qL3Yu69N4kLXM+Xh)`*_*WKu_`QQji2t5k@Q0DJ< zv$L!`29h)CPRncUdCy}zgjnPKsq)a5HT~jw+w^slsIC28vw+~MH<|mJ!2$(Uu}*Hn zb)6G<=K_zU@4jusrpt5=ldWbcm46U?S(qKJ>f>+wXF|RHl4p*gM%O1^z3)!)q*CDr zF9YG?osk-cnu_LaP78GFR;^cNmCevBMS8A?A-Dg!RD3Fc8l&oyD$SOZhXJY}tf^i? zTAJ42qb?h*XA)eKW8hRnDfMlQho@(32qpGCYSGt!P`39t#TdFwvo$D+CV19}7-Fxo^X<&UY_}J^X8L>J8lKC>ZW_ zLzq^_p&z26C_Eyia2O&neaDMlR8>_0fn#A|(QSkm78mrW%1U6S(S79^;|qED?sB8H zxjm(4&(5!|%-RF}W*gn5rKQEQzcv3JeZ?ea$2T8*wX@RqaiPi6cDX0|{X^cA!8P!T zB%;&f2xt+-!^1PIH1C|Qg3Zm&0y$JuUG1{aF6Bl@O-f2iM#e-M{bO)YRYgTj;*-@g zMMWT`jT_za>gy+cV;SW7y?u6PU$yyr>FA8F4kX?k7s>=34zK|O22;*D*vz{YCHz!jn`PEfb{6L$tvMO@zqVNO2{>jbFZEtU{uC7i)Lo+io1G=TG ztbBttjYN9>NP3JHE?2B$+T@Yn)U<~`Z=c~}kJ_<$gdQ_WTtrw>QW8T`ve4r55u^;~ zar1CYv`%2%yLa!@8bVwb+G$8gew|5Ic68iEXd;6i+ve z4V1|jEXMnIO7+(`P0d3?LvT18C>M_VczIeB#Ke37r`!3esTmq8%{L&-Jb;34@!rXh z3AuHwXcKlAFIrz;@9L66wsQr1!?qF5D<2uWWEHG|#p*LLSm95ug$ujAeoVZKC9#fK zp(q#C-7j6KV9yS_9A>q9}I&WZT=s?{|^*;gHVlKt11)n^m>F z-@52-9_BKNH{!z_Io6Tf76OxbU-J4AZQ4b6?O-8=#PvwN)j6yyVUnxr}SMR#KD0-;G}1O z9v*decNd(U+IL|r9+;1_`(D~dy?a9!g5kO+&5DE1M0fN3;r{IvQS0v+T;A4wMB1v= z-@lMa!<|;pk3Kz4rN`{_b$>(AzmnG_C88q><-LWALwQZZ>Kv6fW2DFpLn{>e~Sx%aoA-<@3$ z&+hAs)ryIBEN4by#Q8y`7w2rVLdneSGllr(xvyI<&M$09=T@zz>tiL=HCR!)rJW4M zlC=eV2x|J&?hiCtx|j>W636L$LTq2kVs!GdvxC%FlehjhsWv|c&Q#vK~;sdIkm=v(RI z*3vn8&xml=Qq;C|=@v-}=Z2U;)M#gw82{uKBn^?Qa#{_ah3_r9#r7 zy{TpiKT*&H$F>KTyC#kP$=muHYn?cV(P~yi%0?{?^s;ROoeHjDyThmEv^~1uXtN-F zNO6dvXZrKUW(x)A)!WGUxFB;o8Z`8nhkSlD;8y?;xMCqF1=fZifbz+KoBO5T#-B}QRbxTH879ytChU$&b_ek*B~o(z4v zmakT0gir(h8Ek5F7+rqLKi0sEl4tk3QK=hw1i3M=xOR!(YSks$?wN%h39aHX1p&9f z+rH45d-Ctg5Re*@2=W;m^dWKzLqUw21ru~anH4Fhy*%e9&e=iinTL}Irr3cJ5+qCP z%=0j(Feoi9XsE`A=*EJ zxFVs#k$DQ#Z=kH;+81LH1cl)$hB4>AlU0_?zf(`vlYb}1EVh5AH;pvr1l%6(v-o@w zUv*5?rLCY4-q6snh5-Ztz!snFFTZ@076{TU z*K_?{JMg;xp9U$NpfFEOaP+Fn67ww3E0uCvlr=a1q{A^+ZH=#vkSj_NtNm+(sj+0fzP_~$4TXh;Fu2JN2|fM6@o}AZpJV*6;LFR)+uKf{^I7&i zm;2pB6ApC;sB8%zh&cdALr50Yl-c&6md3_3zu)uo!I#GWl2?)`$BsVP(B>VF7vHol9Ja*O_{nWmB9(f z;MyULgtv~aZqU_E!w?04d->Y-;_8l$oFDs+4p3+8e(vu4>PW!wfU`ZC0D)M`&dyFr z1b~w8k#J96U&zUXz721FVc`=f(2ziBp-?DqL+Ku6kM|SRWo1sNvHYX;pQRNQuFlSL z&0bq2x}|{LY}bu>2p~Yv0Eu~fd;8C$ieD@g4Rc}y7EzDMSk)E}d;_KbaXQ3nd-c!zKG1-Q zL0MVZ(8vhgc0j!6^z5C(D_z|OG&IhJ=zTpsMFiW`V`zU2W|$4?93H46C46?bk9702 zd`~uYk+5Yz>TPUQPMU9_7@)+BRg`TL}J!!<-diDW8LL@nW!M0hld0N!cfsOh33gvpal70fh!jmaA zV|E$eDk^0B4p+BlYJRdJAYY(_jfTUyGBUip6;)L)fN%#3NV2{ue|a382N+>EiJVR2 zR8%-I)xCd76ai3}krr^0Wj+GFN96`kL8;7wIeRLDeYZl?45yyxrRvjeY@@s#yA6#< zw)#CK8Ekcuv!kwNH20QW>pk=>i;W@tExevr5!|+<$>`XKZ>iaY*~S`po-*BSg+J*x z2L5xk_r|p5csNV!2*fVX)S!$3D6p8A7^hwtsKr43c?7^-3+`m&`4$<;GP+epJAiBb zPK-;&rl%zwz$Cc1xBx>GuvuKlOgd2Z>R-x8L@jDAi5u9Me!n+Ie~i8!K{Xw`U^UXs z&cpNJbJ5vC$b!=kuyV07GsmbA&f0dy9j%6YVJ>qGW?k>NhB(^WS9uU-yn=5GX9oas z3+f2@+Ue=(_Wt%t1_lQH{{Fy9z%|41ALfMI`mLyfrT9mGzn#7PFFQeCji&U)DjZW@ zJ|c6Jhnaz4X|~Bz*c4M5otz)6K%vkYo1b_% zI0hiC7|U0EfQJc@y!`tG6BE;+&i{93qhj4Rw+5PDwTJ;1VC zlHcS-h6+_WR=J<`%*_S$#!$X`^{SU2WP2MUxwbBp$hkQaZS88XM{VSW%-7yMi+?U? zJDgQk#s%x>Q-j<7%&^{DlXVIgs=R4eK6gp=<1E8yR4i1(PAVntUHe^m5$O*ngfz^vi=>wN>Bd8l_->(b0@}7TNHoJ!=_NXj0 z@``T^%!{7|S^ThwHH)uct;Yei*=#%(2X5fEz-ZdECeN&*Qf+pdJ3KY zMnRdh=izV1iWm3m0B#3mg!j)rgPIw)QN5iY)q1Z+0aQaxZ5;I)+_6b6FDfPmfOZ#q zdkdEfGFsZ4^z@7ImzrSo{Q?3iOj`6Rrbg1AMB0bDI66xBAO8)*CF`+tfBpJ6g7A^| z&Tk>}&bY+?gkIUL$Km?6{*7|y|W9#YauC1;zv$37|aadbgE?*pN2vLFZp1AvR zPfs*iEHAfl<9;Mvu(`McCp)`Ay;J%@KCqQe%Kv!2Er2Svps~>xgfFPo)ZZr%6Xr#q zCGApFv&U-f#He1%+@>`D<1_I4je9)J(Hp`CIPMf!|IIro%3CcuG3y0NE$z@kDn={Ui&q^9GIHpo zfQ_h(Ovuj+AyBye$jy-YedXXj-uDb4$F!UIrRNnlyxd?H_1kMZOLNpBk&6NJ0Z;f*T73^Hf=^2S>9^f`Bp#0+Ja`~;bBv58W80=R`M+vK|DS)SpDxB} zKA_yc2z>V$fw_1nE)u^oxv{jFTrK0#WXu?kgCTV%%Ki>W8L$H^es%LZIP*$?fB*v} zo%rWpwTs6cgcM$&g(8s1myGdtb-$aN#mn{ZH84^ihM2h(Kt9q+WsGc0O-(rpQXvm4 ztH3tq| zv^Tf}dEz`gX*8X#A6t&hiWqdn4KcfiJ)M%H%J=Ul0aO2e3qUuLlAOc&-}a7mhVMRu zVpzEvkd?|Wl02!EyVmN$I(Qo%Riz1YR2j@}lv-A$*`GjdLkAu4LKc>u$X#Ut)HII; zRtoX#;X~0pjY3cvM*nyHpY2plicGPJD{E^ki34GB3?KW*a4>dvcLmai78Y`qD?GO$ zkCqOX_wc)kM#5=LeR%|P@ON~e!cy9sg5`SXU29%0j1P@+N&b%t;Qzy`X!VzOIxZ7m VeV+QNfG$Fm6*QiGlQVzwzW}aRtnvT= literal 0 HcmV?d00001 diff --git a/tests/v3/test_transport_bindings.py b/tests/v3/test_transport_bindings.py index 006ea33f6..de6fa18c0 100644 --- a/tests/v3/test_transport_bindings.py +++ b/tests/v3/test_transport_bindings.py @@ -197,3 +197,252 @@ def test_load_time_binding_labels_on_load(v3_system: SystemFixture, make_plugin, # The encoder is bound to the Tempo parameter through the load-time row. assert enc1.parameter is tp.parameters[BPM_SYMBOL] snapshot("loaded") + + +def test_encoder_bpm_turn_sends_websocket(v3_system: SystemFixture, make_plugin): + """MIDI Learning /pedalboard :bpm to an encoder and turning it updates local tempo, + shows the dismissible parameter dialog, and dispatches a high-precision WebSocket packet.""" + handler = v3_system.handler + hw = v3_system.hw + ws_bridge = v3_system.ws_bridge + assert handler.current is not None and handler.lcd is not None + + plugin = make_plugin("noise", bypassed=False) + handler.current.pedalboard.plugins = [plugin] + _attach_transport_plugin(handler) + + enc1 = next(e for e in hw.encoders if getattr(e, "id", None) == 1) + channel, cc = _binding_for(hw, enc1).split(":") + + # 1. Live MIDI Learn message from MOD-UI + ws_bridge.inject(f"midi_map /pedalboard :bpm {channel} {cc} 20.0 280.0") + handler.poll_ws_messages() + + tp = handler.current.pedalboard.transport_plugin + assert tp is not None + assert tp.parameters[BPM_SYMBOL].binding == f"{channel}:{cc}" + assert enc1.parameter is tp.parameters[BPM_SYMBOL] + + # 2. Turn encoder clockwise by 1 step (1.0 BPM per detent) + from pistomp.input.event import EncoderEvent + + event = EncoderEvent(controller=enc1, rotations=1, multiplier=1.0) + handler._handle_encoder(event) + + # 3. Tempo parameter updated (1 step clockwise from 120.0 -> 121.0 BPM) + assert tp.parameters[BPM_SYMBOL].value == 121.0 + + # 4. High-precision WebSocket transport-bpm packet queued for MOD-UI + sent_msgs = list(ws_bridge.sent) + assert any("transport-bpm 121.0" in m for m in sent_msgs) + + +def test_encoder_bpm_clamping_at_boundaries(v3_system: SystemFixture, make_plugin): + """BPM parameter edits clamp strictly at minimum (20.0) and maximum (280.0) limits.""" + from pistomp.input.event import EncoderEvent + + handler = v3_system.handler + hw = v3_system.hw + assert handler.current is not None + + plugin = make_plugin("noise", bypassed=False) + handler.current.pedalboard.plugins = [plugin] + + enc1 = next(e for e in hw.encoders if getattr(e, "id", None) == 1) + channel, cc = _binding_for(hw, enc1).split(":") + _attach_transport_plugin( + handler, + bpm_cc={"channel": int(channel), "control": int(cc), "hasRanges": True, "minimum": 20.0, "maximum": 280.0}, + ) + + tp = handler.current.pedalboard.transport_plugin + assert tp is not None + bpm_param = tp.parameters[BPM_SYMBOL] + + # Spin clockwise aggressively (+200 steps) + event_up = EncoderEvent(controller=enc1, rotations=200, multiplier=1.0) + handler._handle_encoder(event_up) + assert bpm_param.value == 280.0 + + # Spin counter-clockwise aggressively (-300 steps) + event_down = EncoderEvent(controller=enc1, rotations=-300, multiplier=1.0) + handler._handle_encoder(event_down) + assert bpm_param.value == 20.0 + + +def test_unbound_encoder_emits_midi_cc_for_learning(v3_system: SystemFixture, make_plugin): + """An unbound encoder emits fallback MIDI CC so MOD-UI can learn it, but once bound + to /pedalboard :bpm, turns bypass MIDI CC and send WebSocket messages instead.""" + from unittest.mock import MagicMock + from pistomp.input.event import EncoderEvent + + handler = v3_system.handler + hw = v3_system.hw + ws_bridge = v3_system.ws_bridge + assert handler.current is not None + + plugin = make_plugin("noise", bypassed=False) + handler.current.pedalboard.plugins = [plugin] + _attach_transport_plugin(handler) + + enc1 = next(e for e in hw.encoders if getattr(e, "id", None) == 1) + channel, cc = _binding_for(hw, enc1).split(":") + + # Mock _emit_midi to observe MIDI CC output + handler._emit_midi = MagicMock() + + # 1. Turn unbound encoder -> emits MIDI CC for MOD-UI MIDI Learn + assert enc1.parameter is None + handler._handle_encoder(EncoderEvent(controller=enc1, rotations=1, multiplier=1.0)) + handler._emit_midi.assert_called_once() + handler._emit_midi.reset_mock() + + # 2. Bind encoder to /pedalboard :bpm via MIDI Learn + ws_bridge.inject(f"midi_map /pedalboard :bpm {channel} {cc} 20.0 280.0") + handler.poll_ws_messages() + assert enc1.parameter is not None + + # 3. Turn bound encoder -> bypasses _emit_midi and sends WebSocket transport-bpm + ws_bridge.sent.clear() + handler._handle_encoder(EncoderEvent(controller=enc1, rotations=1, multiplier=1.0)) + handler._emit_midi.assert_not_called() + assert any("transport-bpm" in msg for msg in ws_bridge.sent) + + +def test_parameter_value_commit_for_transport_bpm(v3_system: SystemFixture, make_plugin): + """parameter_value_commit for /pedalboard :bpm routes via set_mod_tap_tempo (WebSocket).""" + handler = v3_system.handler + ws_bridge = v3_system.ws_bridge + assert handler.current is not None + + plugin = make_plugin("noise", bypassed=False) + handler.current.pedalboard.plugins = [plugin] + _attach_transport_plugin(handler) + + tp = handler.current.pedalboard.transport_plugin + assert tp is not None + bpm_param = tp.parameters[BPM_SYMBOL] + + ws_bridge.sent.clear() + handler.parameter_value_commit(bpm_param, 135.0) + + assert bpm_param.value == 135.0 + assert any("transport-bpm 135.0" in m for m in ws_bridge.sent) + + +def test_encoder_bpm_fast_spin_acceleration(v3_system: SystemFixture, make_plugin): + """Slow encoder turn moves 1.0 BPM per detent; fast spin accelerates edit proportionally.""" + from pistomp.input.event import EncoderEvent + + handler = v3_system.handler + hw = v3_system.hw + assert handler.current is not None + + plugin = make_plugin("noise", bypassed=False) + handler.current.pedalboard.plugins = [plugin] + + enc1 = next(e for e in hw.encoders if getattr(e, "id", None) == 1) + channel, cc = _binding_for(hw, enc1).split(":") + _attach_transport_plugin( + handler, + bpm_cc={"channel": int(channel), "control": int(cc), "hasRanges": True, "minimum": 20.0, "maximum": 280.0}, + ) + + tp = handler.current.pedalboard.transport_plugin + assert tp is not None + bpm_param = tp.parameters[BPM_SYMBOL] + + # 1. Slow turn (multiplier=1.0) -> moves exactly 1.0 BPM (120.0 -> 121.0) + handler._handle_encoder(EncoderEvent(controller=enc1, rotations=1, multiplier=1.0)) + assert bpm_param.value == 121.0 + + # 2. Fast spin (multiplier=4.0) -> accelerates edit (>1.0 BPM) + handler._handle_encoder(EncoderEvent(controller=enc1, rotations=1, multiplier=4.0)) + assert bpm_param.value > 122.0 # Accelerated step (>1 BPM) + + +def test_incoming_transport_decimal_bpm_sync(v3_system: SystemFixture, make_plugin): + """External transport updates (e.g. Ableton Link) preserve decimal BPM values.""" + handler = v3_system.handler + ws_bridge = v3_system.ws_bridge + assert handler.current is not None + + plugin = make_plugin("noise", bypassed=False) + handler.current.pedalboard.plugins = [plugin] + _attach_transport_plugin(handler) + + tp = handler.current.pedalboard.transport_plugin + assert tp is not None + + # Inject external transport frame with decimal BPM (120.5) + ws_bridge.inject("transport 1 4.0 120.5 link") + handler.poll_ws_messages() + + assert tp.parameters[BPM_SYMBOL].value == 120.5 + + +def test_encoder_bpm_turn_without_websocket_bridge_falls_back_to_rest_post( + v3_system: SystemFixture, make_plugin +): + """If ws_bridge.send_bpm returns False, encoder tempo turns execute REST POST fallback.""" + from unittest.mock import MagicMock + from pistomp.input.event import EncoderEvent + + handler = v3_system.handler + hw = v3_system.hw + ws_bridge = v3_system.ws_bridge + mock_post = v3_system.mock_post + assert handler.current is not None + + plugin = make_plugin("noise", bypassed=False) + handler.current.pedalboard.plugins = [plugin] + + enc1 = next(e for e in hw.encoders if getattr(e, "id", None) == 1) + channel, cc = _binding_for(hw, enc1).split(":") + _attach_transport_plugin( + handler, + bpm_cc={"channel": int(channel), "control": int(cc), "hasRanges": True, "minimum": 20.0, "maximum": 280.0}, + ) + + # Mock send_bpm to return False (simulating backpressure/send failure) + ws_bridge.send_bpm = MagicMock(return_value=False) + mock_post.reset_mock() + + # Turn encoder + handler._handle_encoder(EncoderEvent(controller=enc1, rotations=1, multiplier=1.0)) + + # Assert REST POST fallback was executed + mock_post.assert_called_once() + assert "set_bpm" in mock_post.call_args[0][0] + assert mock_post.call_args[1]["json"] == {"value": 121.0} + + +def test_encoder_bpm_turn_parameter_dialog_snapshot(v3_system: SystemFixture, make_plugin, snapshot): + """Turning a BPM-bound encoder 1 detent notch displays the parameter dialog on the LCD at 121 BPM.""" + from pistomp.input.event import EncoderEvent + + handler = v3_system.handler + hw = v3_system.hw + ws_bridge = v3_system.ws_bridge + assert handler.current is not None and handler.lcd is not None + + plugin = make_plugin("noise", bypassed=False) + handler.current.pedalboard.plugins = [plugin] + _attach_transport_plugin(handler) + + handler.lcd.link_data(handler.pedalboard_list, handler.current, hw.footswitches) + handler.lcd.draw_main_panel() + + enc1 = next(e for e in hw.encoders if getattr(e, "id", None) == 1) + channel, cc = _binding_for(hw, enc1).split(":") + + ws_bridge.inject(f"midi_map /pedalboard :bpm {channel} {cc} 20.0 280.0") + handler.poll_ws_messages() + + # Turn encoder 1 detent (120.0 -> 121.0 BPM) + handler._handle_encoder(EncoderEvent(controller=enc1, rotations=1, multiplier=1.0)) + + # Capture LCD snapshot of 121 BPM parameter dialog badge + snapshot("bpm_dialog_121") + + From a38eaba961f1c1a422b29011a69f77be6b092d22 Mon Sep 17 00:00:00 2001 From: John Remillard Date: Wed, 22 Jul 2026 18:54:00 -0400 Subject: [PATCH 2/4] fixed linting issue --- plugins/transport/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/transport/__init__.py b/plugins/transport/__init__.py index ef376e81b..46f210ea0 100644 --- a/plugins/transport/__init__.py +++ b/plugins/transport/__init__.py @@ -9,7 +9,7 @@ from __future__ import annotations -from common.parameter import Parameter, Symbol +from common.parameter import Parameter from modalapi.plugin_customization import PluginCustomization from modalapi.pedalboard import BPM_SYMBOL, BPB_SYMBOL, ROLLING_SYMBOL from plugins.customization import register From 511a153636b45288e475973edeef5ccbdd8013bb Mon Sep 17 00:00:00 2001 From: John Remillard Date: Wed, 22 Jul 2026 19:17:14 -0400 Subject: [PATCH 3/4] addressing comments from feedback --- modalapi/modhandler.py | 45 ++++++++------ modalapi/pedalboard.py | 57 +++++++----------- tests/v3/test_transport_bindings.py | 93 +++++++++++++++++++++++++++++ 3 files changed, 140 insertions(+), 55 deletions(-) diff --git a/modalapi/modhandler.py b/modalapi/modhandler.py index 9cddd3f76..5e9b12d4d 100755 --- a/modalapi/modhandler.py +++ b/modalapi/modhandler.py @@ -225,6 +225,10 @@ def __init__(self, audiocard: Audiocard, homedir, data_dir="/home/pistomp/data") # Suppress outbound WebSocket messages while a pedalboard change is in flight. self._is_pedalboard_loading = False + # Reactive BPM parameter observer state + self._bpm_unsub: Callable[[], None] | None = None + self._suppress_bpm_event: bool = False + # Tuner state self._tuner_source_factory: TunerSourceFactory | None = None self._tuner_source_spec: str = "jack" @@ -419,8 +423,6 @@ def _handle_encoder(self, event: EncoderEvent) -> bool: new_value = ParameterSteps.for_parameter(c.parameter).move(delta) c.parameter.value = new_value self.lcd.display_parameter_value(c.parameter, new_value) - if c.parameter.instance_id == Pedalboard.TRANSPORT_INSTANCE_ID and c.parameter.symbol == BPM_SYMBOL: - self.set_mod_tap_tempo(new_value) emit_value = c.bar_midi_value() else: emit_value = self._advance_encoder_fallback(c, delta) @@ -428,10 +430,8 @@ def _handle_encoder(self, event: EncoderEvent) -> bool: # Unconditional, and must stay that way: an unbound encoder has no row, # and this emit is the only way mod-ui sees its CC to MIDI-learn it. # Emission is hardware-level, below the table (see input/README.md). - # Transport BPM parameters bypass 7-bit MIDI CC emission for high-precision WebSocket transport. - if c.parameter is None or not ( - c.parameter.instance_id == Pedalboard.TRANSPORT_INSTANCE_ID and c.parameter.symbol == BPM_SYMBOL - ): + # Transport parameters bypass 7-bit MIDI CC emission for high-precision WebSocket transport. + if c.parameter is None or c.parameter.instance_id != Pedalboard.TRANSPORT_INSTANCE_ID: self._emit_midi(c, emit_value) return True @@ -881,13 +881,13 @@ def _handle_ws_message(self, msg: WebSocketMessage): # MIDI slave, another HMI). :rolling's enum flips Playing/Stopped. if self._current is not None: tp = self.current.pedalboard.transport_plugin - if tp is not None: - if ROLLING_SYMBOL in tp.parameters: - tp.set_param_value(ROLLING_SYMBOL, 1.0 if msg.rolling else 0.0) - if BPM_SYMBOL in tp.parameters: - tp.set_param_value(BPM_SYMBOL, msg.bpm) - if BPB_SYMBOL in tp.parameters: - tp.set_param_value(BPB_SYMBOL, msg.beats_per_bar) + tp.set_param_value(ROLLING_SYMBOL, 1.0 if msg.rolling else 0.0) + tp.set_param_value(BPB_SYMBOL, msg.beats_per_bar) + self._suppress_bpm_event = True + try: + tp.set_param_value(BPM_SYMBOL, msg.bpm) + finally: + self._suppress_bpm_event = False if self.hardware and self.hardware.taptempo: self.hardware.taptempo.set_bpm(msg.bpm) if self.hardware.taptempo.is_enabled(): @@ -1213,6 +1213,19 @@ def bind_current_pedalboard(self): # The pedalboard data has already been loaded, but this will overlay # any real time settings self._controller_manager.bind(self.current) + self._bind_transport_bpm_listener() + + def _bind_transport_bpm_listener(self) -> None: + if self._bpm_unsub is not None: + self._bpm_unsub() + self._bpm_unsub = None + if self._current is not None: + bpm_param = self.current.pedalboard.transport_plugin.parameters[BPM_SYMBOL] + self._bpm_unsub = bpm_param.subscribe(self._on_bpm_param_changed) + + def _on_bpm_param_changed(self, param: Parameter) -> None: + if not self._suppress_bpm_event: + self.set_mod_tap_tempo(param.value) def _redraw_after_binding(self, controller: Controller, is_footswitch: bool) -> None: if is_footswitch: @@ -1433,10 +1446,6 @@ def parameter_value_commit(self, param, value): # Audio parameter (volume, EQ, etc.) - handled locally, no remote update needed if param.instance_id is None: self.audio_parameter_commit(param.symbol, value) - # Pedalboard-level transport BPM parameters are set via WebSocket for precision - if param.instance_id == Pedalboard.TRANSPORT_INSTANCE_ID and param.symbol == BPM_SYMBOL: - self.set_mod_tap_tempo(value) - return # External MIDI parameters have no mod-host counterpart. The dialog's NAV # path owns sending the CC that _handle_encoder would have sent for a turn; @@ -1451,7 +1460,7 @@ def parameter_value_commit(self, param, value): self._emit_midi(controller, int(value)) return - if not self._is_pedalboard_loading and param.instance_id is not None: + if not self._is_pedalboard_loading and param.instance_id is not None and param.instance_id != Pedalboard.TRANSPORT_INSTANCE_ID: self.ws_bridge.send_parameter(param.instance_id, param.symbol, param.value) @property diff --git a/modalapi/pedalboard.py b/modalapi/pedalboard.py index b9cd080cf..dc8127408 100755 --- a/modalapi/pedalboard.py +++ b/modalapi/pedalboard.py @@ -102,10 +102,8 @@ def __init__(self, title, bundle, root_uri="http://localhost:80/", customizer: C self.connections: list[Connection] = [] self.hydrated = False # Synthetic /pedalboard pseudo-instance carrying :bpm/:bpb/:rolling. - # Built in hydrate() from timeInfo; None before the first hydrate and - # on boards with no transport metadata. Excluded from self.plugins so - # the effect-graph render never paints it. - self.transport_plugin: Plugin.Plugin | None = None + # Excluded from self.plugins so the effect-graph render never paints it. + self.transport_plugin: Plugin.Plugin = self._build_transport_plugin(None) def get_plugin_data(self, uri): url = self.root_uri + "effect/get?uri=" + urllib.parse.quote(uri) @@ -238,45 +236,31 @@ def hydrate(self, plugin_dict) -> None: self.hydrated = True - def _build_transport_plugin(self, time_info: dict | None) -> Plugin.Plugin | None: + def _build_transport_plugin(self, time_info: dict | None) -> Plugin.Plugin: """The /pedalboard pseudo-instance carrying :bpm/:bpb/:rolling. Built - from mod-ui's timeInfo block. None when the board reports no transport - metadata (available == 0 or absent).""" - if not time_info: - return None - available = int(time_info.get("available", 0) or 0) - if available == 0: - return None + from mod-ui's timeInfo block (or default unbound parameters when absent).""" + time_info = time_info or {} - parameters: dict[Symbol, Parameter] = {} - # mod-ui's kPedalboardTimeAvailable* bit masks. - if available & 0x1: # BPB - cc = time_info.get("bpbCC") - parameters[BPB_SYMBOL] = Parameter( + parameters: dict[Symbol, Parameter] = { + BPB_SYMBOL: Parameter( _transport_port_info(BPB_SYMBOL), - float(time_info.get("bpb", _BPB_RANGE[0])), - self._binding(cc), + float(time_info.get("bpb", 4.0)), + self._binding(time_info.get("bpbCC")), TRANSPORT_INSTANCE_ID, - ) - if available & 0x2: # BPM - cc = time_info.get("bpmCC") - parameters[BPM_SYMBOL] = Parameter( + ), + BPM_SYMBOL: Parameter( _transport_port_info(BPM_SYMBOL), - float(time_info.get("bpm", _BPM_RANGE[0])), - self._binding(cc), + float(time_info.get("bpm", 120.0)), + self._binding(time_info.get("bpmCC")), TRANSPORT_INSTANCE_ID, - ) - if available & 0x4: # Rolling - cc = time_info.get("rollingCC") - parameters[ROLLING_SYMBOL] = Parameter( + ), + ROLLING_SYMBOL: Parameter( _transport_port_info(ROLLING_SYMBOL), 1.0 if time_info.get("rolling") else 0.0, - self._binding(cc), + self._binding(time_info.get("rollingCC")), TRANSPORT_INSTANCE_ID, - ) - - if not parameters: - return None + ), + } # category drives footswitch color; "Utility" is the benign choice for # transport — no LV2 category exists for it. uri=urn:mod:pedalboard so @@ -296,9 +280,8 @@ def find_plugin(self, instance_id: str) -> Plugin.Plugin | None: for p in self.plugins: if p.instance_id == instance_id: return p - tp = self.transport_plugin - if tp is not None and tp.instance_id == instance_id: - return tp + if self.transport_plugin.instance_id == instance_id: + return self.transport_plugin return None def _build_plugin(self, instance_id: str, uri: str, x: float, y: float, info: dict) -> Optional[Plugin.Plugin]: diff --git a/tests/v3/test_transport_bindings.py b/tests/v3/test_transport_bindings.py index de6fa18c0..618365821 100644 --- a/tests/v3/test_transport_bindings.py +++ b/tests/v3/test_transport_bindings.py @@ -64,6 +64,99 @@ def test_transport_plugin_built_from_timeinfo(v3_system: SystemFixture): assert rolling.value == 0.0 +def test_transport_plugin_unconditional_fallback(v3_system: SystemFixture): + """When timeInfo is empty, None, or available=0, transport_plugin is unconditionally + created with default unbound parameters for :bpm, :bpb, and :rolling.""" + handler = v3_system.handler + assert handler.current is not None + + # Test empty dict/None + tp = handler.current.pedalboard._build_transport_plugin(None) + assert tp is not None + assert tp.instance_id == TRANSPORT_INSTANCE_ID + assert BPM_SYMBOL in tp.parameters + assert BPB_SYMBOL in tp.parameters + assert ROLLING_SYMBOL in tp.parameters + assert tp.parameters[BPM_SYMBOL].binding is None + assert tp.parameters[BPM_SYMBOL].value == 120.0 + assert tp.parameters[BPB_SYMBOL].value == 4.0 + assert tp.parameters[ROLLING_SYMBOL].value == 0.0 + + # Test available=0 + tp_zero = handler.current.pedalboard._build_transport_plugin({"available": 0}) + assert tp_zero is not None + assert BPM_SYMBOL in tp_zero.parameters + assert BPB_SYMBOL in tp_zero.parameters + assert ROLLING_SYMBOL in tp_zero.parameters + + +def test_reactive_bpm_parameter_change_triggers_set_mod_tap_tempo(v3_system: SystemFixture): + """Writing to transport_plugin.parameters[BPM_SYMBOL].value reactively notifies + subscribers and triggers set_mod_tap_tempo.""" + from unittest.mock import MagicMock + + handler = v3_system.handler + ws_bridge = v3_system.ws_bridge + assert handler.current is not None + + _attach_transport_plugin(handler) + ws_bridge.send_bpm = MagicMock(return_value=True) + + # Change BPM parameter value directly (e.g. via encoder or set_param_value) + tp = handler.current.pedalboard.transport_plugin + tp.set_param_value(BPM_SYMBOL, 148.0) + + # Verify reactive subscriber triggered send_bpm + ws_bridge.send_bpm.assert_called_once_with(148.0) + + +def test_transport_message_ws_suppresses_bpm_echo(v3_system: SystemFixture): + """An incoming WebSocket TransportMessage updates transport parameters without + echo-calling send_bpm back to mod-ui.""" + from unittest.mock import MagicMock + + handler = v3_system.handler + ws_bridge = v3_system.ws_bridge + assert handler.current is not None + + _attach_transport_plugin(handler) + ws_bridge.send_bpm = MagicMock(return_value=True) + + # Inject incoming WS TransportMessage from mod-ui: transport {rolling} {bpb} {bpm} {syncMode} + ws_bridge.inject("transport 1 4.0 155.0 Internal") + handler.poll_ws_messages() + + tp = handler.current.pedalboard.transport_plugin + assert tp.parameters[BPM_SYMBOL].value == 155.0 + assert tp.parameters[ROLLING_SYMBOL].value == 1.0 + assert tp.parameters[BPB_SYMBOL].value == 4.0 + + # Ensure send_bpm was NOT called because update originated from WS + ws_bridge.send_bpm.assert_not_called() + + +def test_partial_timeinfo_bitmask_creates_all_parameters(v3_system: SystemFixture): + """When timeInfo has a partial bitmask (e.g. only BPM), all 3 parameters + are still created, with missing ones remaining unbound.""" + handler = v3_system.handler + assert handler.current is not None + + # Available bitmask 0x2 = BPM only + tp = handler.current.pedalboard._build_transport_plugin({ + "available": 0x2, + "bpm": 130.0, + "bpmCC": {"channel": 0, "control": 10}, + }) + assert tp is not None + assert BPM_SYMBOL in tp.parameters + assert BPB_SYMBOL in tp.parameters + assert ROLLING_SYMBOL in tp.parameters + + assert tp.parameters[BPM_SYMBOL].binding == "0:10" + assert tp.parameters[BPB_SYMBOL].binding is None + assert tp.parameters[ROLLING_SYMBOL].binding is None + + def test_find_plugin_resolves_transport(v3_system: SystemFixture): """find_plugin returns the transport pseudo-plugin for the /pedalboard id, where the old `next(p for p in plugins)` lookup returned None.""" From e1ec1b3b8560958b265f12070c23a20ebb193871 Mon Sep 17 00:00:00 2001 From: John Remillard Date: Fri, 24 Jul 2026 01:25:45 -0400 Subject: [PATCH 4/4] refactor: address PR #216 review feedback (symbol-level MIDI CC bypass, audio param early return, and test enhancements) --- modalapi/modhandler.py | 15 ++++--- tests/integration/test_tap_tempo.py | 15 ++++--- tests/v3/test_transport_bindings.py | 61 +++++++++++++++++++++++++++-- 3 files changed, 77 insertions(+), 14 deletions(-) diff --git a/modalapi/modhandler.py b/modalapi/modhandler.py index 5e9b12d4d..687ad1d52 100755 --- a/modalapi/modhandler.py +++ b/modalapi/modhandler.py @@ -431,7 +431,9 @@ def _handle_encoder(self, event: EncoderEvent) -> bool: # and this emit is the only way mod-ui sees its CC to MIDI-learn it. # Emission is hardware-level, below the table (see input/README.md). # Transport parameters bypass 7-bit MIDI CC emission for high-precision WebSocket transport. - if c.parameter is None or c.parameter.instance_id != Pedalboard.TRANSPORT_INSTANCE_ID: + if c.parameter is None or not ( + c.parameter.instance_id == Pedalboard.TRANSPORT_INSTANCE_ID and c.parameter.symbol == BPM_SYMBOL + ): self._emit_midi(c, emit_value) return True @@ -1446,6 +1448,7 @@ def parameter_value_commit(self, param, value): # Audio parameter (volume, EQ, etc.) - handled locally, no remote update needed if param.instance_id is None: self.audio_parameter_commit(param.symbol, value) + return # External MIDI parameters have no mod-host counterpart. The dialog's NAV # path owns sending the CC that _handle_encoder would have sent for a turn; @@ -1460,7 +1463,9 @@ def parameter_value_commit(self, param, value): self._emit_midi(controller, int(value)) return - if not self._is_pedalboard_loading and param.instance_id is not None and param.instance_id != Pedalboard.TRANSPORT_INSTANCE_ID: + if not self._is_pedalboard_loading and param.instance_id is not None and not ( + param.instance_id == Pedalboard.TRANSPORT_INSTANCE_ID and param.symbol == BPM_SYMBOL + ): self.ws_bridge.send_parameter(param.instance_id, param.symbol, param.value) @property @@ -1818,10 +1823,10 @@ def get_callback(self, callback_name): def set_mod_tap_tempo(self, bpm: float | None) -> None: if bpm is not None: - self._last_bpm_change_time = time.time() - if self.ws_bridge is not None: + if self._ws_bridge is not None: self.ws_bridge.send_bpm(bpm) - self._rest_post(self.root_uri + "set_bpm", json={"value": bpm}) + else: + self._rest_post(self.root_uri + "set_bpm", json={"value": bpm}) def set_sync_mode(self, mode: SyncMode) -> None: """Optimistically switch the clock source; mod-ui's transport echo diff --git a/tests/integration/test_tap_tempo.py b/tests/integration/test_tap_tempo.py index 2702864e7..8643a186d 100644 --- a/tests/integration/test_tap_tempo.py +++ b/tests/integration/test_tap_tempo.py @@ -6,16 +6,21 @@ def test_set_mod_tap_tempo(modhandler_system: SystemFixture): - """set_mod_tap_tempo() POSTs to /set_bpm with the BPM value.""" + """set_mod_tap_tempo() sends high-precision BPM via WS bridge (or REST fallback).""" handler = modhandler_system.handler + ws_bridge = modhandler_system.ws_bridge mock_post = modhandler_system.mock_post handler.set_mod_tap_tempo(120) - mock_post.assert_called_once() - call_args = mock_post.call_args - assert "set_bpm" in call_args.args[0] - assert call_args.kwargs.get("json", {}).get("value") == 120 + if handler._ws_bridge is not None: + assert any("transport-bpm 120" in msg for msg in ws_bridge.sent) + mock_post.assert_not_called() + else: + mock_post.assert_called_once() + call_args = mock_post.call_args + assert "set_bpm" in call_args.args[0] + assert call_args.kwargs.get("json", {}).get("value") == 120 def test_set_mod_tap_tempo_none(modhandler_system: SystemFixture): diff --git a/tests/v3/test_transport_bindings.py b/tests/v3/test_transport_bindings.py index 618365821..eca5afd4e 100644 --- a/tests/v3/test_transport_bindings.py +++ b/tests/v3/test_transport_bindings.py @@ -477,13 +477,12 @@ def test_incoming_transport_decimal_bpm_sync(v3_system: SystemFixture, make_plug def test_encoder_bpm_turn_without_websocket_bridge_falls_back_to_rest_post( v3_system: SystemFixture, make_plugin ): - """If ws_bridge.send_bpm returns False, encoder tempo turns execute REST POST fallback.""" + """If ws_bridge is None, encoder tempo turns execute REST POST fallback.""" from unittest.mock import MagicMock from pistomp.input.event import EncoderEvent handler = v3_system.handler hw = v3_system.hw - ws_bridge = v3_system.ws_bridge mock_post = v3_system.mock_post assert handler.current is not None @@ -497,8 +496,8 @@ def test_encoder_bpm_turn_without_websocket_bridge_falls_back_to_rest_post( bpm_cc={"channel": int(channel), "control": int(cc), "hasRanges": True, "minimum": 20.0, "maximum": 280.0}, ) - # Mock send_bpm to return False (simulating backpressure/send failure) - ws_bridge.send_bpm = MagicMock(return_value=False) + # Set _ws_bridge to None to simulate missing WebSocket bridge + handler._ws_bridge = None mock_post.reset_mock() # Turn encoder @@ -539,3 +538,57 @@ def test_encoder_bpm_turn_parameter_dialog_snapshot(v3_system: SystemFixture, ma snapshot("bpm_dialog_121") +def test_encoder_rolling_or_bpb_emits_midi_cc(v3_system: SystemFixture, make_plugin): + """Transport parameters other than :bpm (e.g., :rolling, :bpb) still emit MIDI CC when turned.""" + from unittest.mock import MagicMock + from modalapi.pedalboard import ROLLING_SYMBOL + from pistomp.input.event import EncoderEvent + + handler = v3_system.handler + hw = v3_system.hw + ws_bridge = v3_system.ws_bridge + assert handler.current is not None + + plugin = make_plugin("noise", bypassed=False) + handler.current.pedalboard.plugins = [plugin] + _attach_transport_plugin(handler) + + enc1 = next(e for e in hw.encoders if getattr(e, "id", None) == 1) + channel, cc = _binding_for(hw, enc1).split(":") + + ws_bridge.inject(f"midi_map /pedalboard :rolling {channel} {cc} 0.0 1.0") + handler.poll_ws_messages() + + tp = handler.current.pedalboard.transport_plugin + assert tp is not None + assert enc1.parameter is tp.parameters[ROLLING_SYMBOL] + + handler._emit_midi = MagicMock() + handler._handle_encoder(EncoderEvent(controller=enc1, rotations=1, multiplier=1.0)) + + # Assert _emit_midi was called for :rolling (unlike :bpm) + handler._emit_midi.assert_called_once() + + +def test_audio_parameter_commit_early_return(v3_system: SystemFixture): + """Audio parameters (instance_id is None) commit locally and return early without sending WS or MIDI CC.""" + from unittest.mock import MagicMock + + handler = v3_system.handler + ws_bridge = v3_system.ws_bridge + assert handler.current is not None + + param = handler._create_audio_parameter("Input Gain", "capture_volume", -19.75, 12) + assert param.instance_id is None + + ws_bridge.send_parameter = MagicMock() + handler._emit_midi = MagicMock() + + handler.parameter_value_commit(param, 0.0) + + # Audio param handled locally, no remote parameter set or MIDI CC emitted + ws_bridge.send_parameter.assert_not_called() + handler._emit_midi.assert_not_called() + + +