Spaces:
Sleeping
Sleeping
| import streamlit as st | |
| import streamlit.components.v1 as components | |
| import subprocess | |
| import time | |
| import os | |
| import threading | |
| from queue import Queue | |
| import json | |
| import mido | |
| # ------------------------------------------------------------ | |
| # Utility Functions | |
| # ------------------------------------------------------------ | |
| def install_fluidsynth(): | |
| """ | |
| Check if 'fluidsynth' CLI is installed. Attempt to install if not found. | |
| On non-Linux systems, you’ll need to manually install fluidsynth. | |
| """ | |
| try: | |
| subprocess.run(['fluidsynth', '--version'], capture_output=True, check=True) | |
| return True | |
| except (FileNotFoundError, subprocess.CalledProcessError): | |
| st.error("FluidSynth not found. Attempting automatic install (Linux only).") | |
| 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 automatically: {str(e)}") | |
| st.code("Please install fluidsynth manually.") | |
| return False | |
| def download_soundfont(sf_filename="GeneralUser_GS_v1.471.sf2"): | |
| """ | |
| Downloads a free SoundFont if it's not already present. | |
| """ | |
| if not os.path.exists(sf_filename): | |
| 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', | |
| sf_filename | |
| ], check=True) | |
| except subprocess.CalledProcessError as e: | |
| st.error(f"Failed to download soundfont: {str(e)}") | |
| return False | |
| return True | |
| # ------------------------------------------------------------ | |
| # FluidSynth Handling Class | |
| # ------------------------------------------------------------ | |
| class FluidSynthPlayer: | |
| """ | |
| Wraps the fluidsynth CLI process, sends commands (like note on/off). | |
| """ | |
| def __init__(self, soundfont_path, gain=2.0, audio_driver='pulseaudio'): | |
| self.soundfont_path = soundfont_path | |
| self.process = None | |
| self.running = False | |
| self.event_queue = Queue() | |
| self.gain = str(gain) | |
| self.audio_driver = audio_driver | |
| def start(self): | |
| """Start FluidSynth as a subprocess.""" | |
| try: | |
| self.process = subprocess.Popen( | |
| [ | |
| 'fluidsynth', | |
| '-a', self.audio_driver, | |
| '-g', self.gain, | |
| 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 the FluidSynth process.""" | |
| self.running = False | |
| if self.process: | |
| self.process.terminate() | |
| self.process.wait() | |
| def _process_events(self): | |
| """Thread to process MIDI events from queue and send to fluidsynth.""" | |
| while self.running: | |
| try: | |
| event = self.event_queue.get(timeout=0.1) | |
| msg_type = event['type'] | |
| if msg_type == 'noteOn': | |
| self._send_command(f"noteon 0 {event['note']} {event['velocity']}") | |
| elif msg_type == 'noteOff': | |
| self._send_command(f"noteoff 0 {event['note']}") | |
| except: | |
| continue | |
| def _send_command(self, command): | |
| """Send a textual command to fluidsynth.""" | |
| 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 a MIDI event to the fluidsynth queue.""" | |
| self.event_queue.put(event) | |
| # ------------------------------------------------------------ | |
| # Basic Arpeggiator | |
| # ------------------------------------------------------------ | |
| class Arpeggiator: | |
| """ | |
| A simple arpeggiator that cycles through held notes at a certain speed. | |
| """ | |
| def __init__(self, synth_player, bpm=120): | |
| self.synth_player = synth_player | |
| self.bpm = bpm | |
| self.notes_held = set() | |
| self.thread = None | |
| self.running = False | |
| def start(self): | |
| if self.running: | |