import streamlit as st
import streamlit.components.v1 as components
import mido
from queue import Queue
import threading
# Store MIDI messages in session state
if 'incoming_midi' not in st.session_state:
st.session_state.incoming_midi = Queue()
def midi_in_callback(msg):
if msg.type in ['note_on', 'note_off']:
st.session_state.incoming_midi.put({
'type': 'noteOn' if msg.type == 'note_on' else 'noteOff',
'note': msg.note,
'velocity': msg.velocity
})
def open_midi_port(port_name: str, is_input: bool):
if port_name == "None":
return None
return mido.open_input(port_name, callback=midi_in_callback) if is_input else mido.open_output(port_name)
def main():
st.title("5-Octave Synth with Arpeggiator & Drum Pads")
# MIDI port selection
ports = {"in": ["None"] + mido.get_input_names(),
"out": ["None"] + mido.get_output_names()}
selections = {
"in": st.selectbox("MIDI Input", ports["in"]),
"out": st.selectbox("MIDI Output", ports["out"])
}
# Initialize/update MIDI ports
for direction in ["in", "out"]:
key = f'midi_{direction}'
if key not in st.session_state:
st.session_state[key] = None
curr_port = st.session_state[key]
if curr_port and curr_port.name != selections[direction]:
curr_port.close()
st.session_state[key] = open_midi_port(selections[direction], direction=="in")
elif not curr_port and selections[direction] != "None":
st.session_state[key] = open_midi_port(selections[direction], direction=="in")
# Load and embed synth interface
components.html(get_synth_interface(), height=800)
# Event handling setup
if 'js_events' not in st.session_state:
st.session_state.js_events = Queue()
if 'browser_events' not in st.session_state:
from collections import deque
st.session_state.browser_events = deque()
# Process hardware MIDI
while not st.session_state.incoming_midi.empty():
evt = st.session_state.incoming_midi.get_nowait()
st.session_state.js_events.put(evt)
# Send events to browser
js_events = []
while not st.session_state.js_events.empty():
js_events.append(st.session_state.js_events.get_nowait())
if js_events:
js_code = ""
components.html(js_code, height=0)
# Cleanup on session end
def cleanup():
for direction in ["in", "out"]:
port = st.session_state.get(f'midi_{direction}')
if port:
port.close()
st.on_session_end(cleanup)
def get_synth_interface():
return """