File size: 4,856 Bytes
75a5b95
 
827e184
90df7ac
 
 
 
 
3648521
 
 
 
 
75a5b95
90df7ac
3648521
 
 
 
827e184
3648521
90df7ac
3648521
 
90df7ac
 
 
 
 
3648521
 
90df7ac
 
3648521
 
 
 
 
90df7ac
 
 
 
 
 
3648521
90df7ac
 
 
 
827e184
 
3648521
 
 
 
90df7ac
3648521
 
 
 
90df7ac
 
 
3648521
 
 
90df7ac
 
3648521
90df7ac
 
 
 
3648521
 
90df7ac
 
 
 
 
 
 
827e184
90df7ac
 
 
 
 
 
 
 
3648521
90df7ac
 
 
 
 
 
3648521
90df7ac
 
 
3648521
 
90df7ac
3648521
90df7ac
 
 
 
 
3648521
90df7ac
827e184
90df7ac
 
 
 
 
 
3648521
90df7ac
75a5b95
3648521
 
 
75a5b95
3648521
75a5b95
3648521
 
 
 
 
 
 
 
75a5b95
3648521
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
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: