import streamlit as st
import streamlit.components.v1 as components
import subprocess
import time
import os
import threading
from queue import Queue
import json
def install_fluidsynth():
"""Check and install FluidSynth if needed"""
try:
subprocess.run(['fluidsynth', '--version'], capture_output=True)
return True
except FileNotFoundError:
st.error("FluidSynth not found. Installing required packages...")
try:
subprocess.run(['sudo', 'apt-get', 'update'], check=True)
subprocess.run(['sudo', 'apt-get', 'install', '-y', 'fluidsynth'], check=True)
return True
except subprocess.CalledProcessError as e:
st.error(f"Failed to install FluidSynth: {str(e)}")
st.code("sudo apt-get install -y fluidsynth")
return False
def download_soundfont():
"""Download a free soundfont if not present"""
soundfont_path = "GeneralUser GS v1.471.sf2"
if not os.path.exists(soundfont_path):
st.info("Downloading soundfont...")
try:
subprocess.run([
'wget',
'https://storage.googleapis.com/google-code-archive-downloads/v2/code.google.com/musicbox/GeneralUser%20GS%20v1.471.sf2',
'-O',
soundfont_path
], check=True)
return True
except subprocess.CalledProcessError as e:
st.error(f"Failed to download soundfont: {str(e)}")
return False
return True
class FluidSynthPlayer:
def __init__(self, soundfont_path):
self.soundfont_path = soundfont_path
self.process = None
self.event_queue = Queue()
self.running = False
def start(self):
"""Start FluidSynth process"""
try:
self.process = subprocess.Popen(
[
'fluidsynth',
'-a', 'pulseaudio', # Use PulseAudio
'-g', '2.0', # Gain (volume)
self.soundfont_path
],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1
)
self.running = True
threading.Thread(target=self._process_events, daemon=True).start()
return True
except Exception as e:
st.error(f"Failed to start FluidSynth: {str(e)}")
return False
def stop(self):
"""Stop FluidSynth process"""
self.running = False
if self.process:
self.process.terminate()
self.process.wait()
def _process_events(self):
"""Process MIDI events from queue"""
while self.running:
try:
event = self.event_queue.get(timeout=0.1)
if event['type'] == 'noteOn':
self._send_command(f"noteon 0 {event['note']} {event['velocity']}")
elif event['type'] == 'noteOff':
self._send_command(f"noteoff 0 {event['note']}")
except:
continue
def _send_command(self, command):
"""Send command to FluidSynth process"""
if self.process and self.process.poll() is None:
try:
self.process.stdin.write(command + '\n')
self.process.stdin.flush()
except:
pass
def queue_event(self, event):
"""Add MIDI event to queue"""
self.event_queue.put(event)
def get_piano_html():
"""Return the HTML content for the piano keyboard"""
return """
"""
def main():
st.title("Piano Keyboard with FluidSynth")
st.write("Click keys or use your computer keyboard (A-K and W-U for white and black keys)")
# Check and install FluidSynth if needed
if not install_fluidsynth():
return
# Download soundfont if needed
if not download_soundfont():
return
# Initialize FluidSynth
if 'synth' not in st.session_state:
st.session_state.synth = FluidSynthPlayer("GeneralUser GS v1.471.sf2")
if not st.session_state.synth.start():
st.error("Failed to start FluidSynth. Please check your audio setup.")
return
# Create a placeholder for messages
message_placeholder = st.empty()
# Display the piano keyboard
components.html(
get_piano_html(),
height=200,
scrolling=False
)
# Handle MIDI events from JavaScript
if 'midi_events' not in st.session_state:
st.session_state.midi_events = []
def handle_midi_event(event):
st.session_state.synth.queue_event(event)
if event['type'] == 'noteOn':
message_placeholder.write(f"Note On: {event['note']}")
else:
message_placeholder.write(f"Note Off: {event['note']}")
# JavaScript callback handler
components.html(
"""
""",
height=0
)
# Cleanup on session end
def cleanup():
if 'synth' in st.session_state:
st.session_state.synth.stop()
st.on_session_ended(cleanup)
if __name__ == "__main__":
main()