import os import numpy as np import soundfile as sf import scipy.signal as signal from app.config import settings from app.core.vst_engine import ( render_midi_events_to_audio, PluginManager, DecentSamplerManager, HAS_PEDALBOARD, HAS_PYFLUIDSYNTH, ) if HAS_PEDALBOARD: try: from pedalboard import Pedalboard, Gain, Chorus, Reverb except Exception: HAS_PEDALBOARD = False class PythonRenderEngine: def __init__(self, sample_rate=44100): self.sample_rate = sample_rate def bars_to_samples(self, bars: float, bpm: float, time_sig_num: int) -> int: seconds_per_beat = 60.0 / max(20.0, bpm) seconds_per_bar = seconds_per_beat * time_sig_num return int(bars * seconds_per_bar * self.sample_rate) def resolve_file_path(self, url_or_id: str) -> str: if not url_or_id: return "" base = os.path.basename(url_or_id) # Check uploads directory p_uploads = os.path.join(settings.UPLOADS_DIR, base) if os.path.exists(p_uploads): return p_uploads # Check processed directory p_processed = os.path.join(settings.PROCESSED_DIR, base) if os.path.exists(p_processed): return p_processed # Check general storage directory p_storage = os.path.join(settings.STORAGE_DIR, base) if os.path.exists(p_storage): return p_storage # Direct check if os.path.exists(url_or_id): return url_or_id return url_or_id def render_session_container(self, session: dict, section_store: dict, bpm: float, time_sig_num: int, total_samples: int) -> np.ndarray: session_buffer = np.zeros((2, total_samples), dtype=np.float32) _channel_counter = 0 for track in session.get("tracks", []): track_type = track.get("type", "AUDIO") track_buffer = np.zeros((2, total_samples), dtype=np.float32) soundfont_bank = track.get("soundfont_bank", 0) soundfont_program = track.get("soundfont_program", 0) is_percussion = track.get("is_percussion", False) or (soundfont_bank == 128) midi_channel = 9 if is_percussion else (_channel_counter % 9) if not is_percussion: _channel_counter += 1 for item in track.get("items", []): start_sample = self.bars_to_samples(item["start_bar"], bpm, time_sig_num) dur_samples = self.bars_to_samples(item["duration_bars"], bpm, time_sig_num) offset_sample = self.bars_to_samples(item["clip_start_offset_bars"], bpm, time_sig_num) item_type = item.get("type") if item_type == "AUDIO_ITEM": source_data = item.get("source_data", {}) audio_url = source_data.get("audio_file_url", "") resolved_path = self.resolve_file_path(audio_url) if resolved_path and os.path.exists(resolved_path): try: audio_data, sr = sf.read(resolved_path, dtype='float32') if sr != self.sample_rate: # Resampling fallback if simple, otherwise skip pass # Handle channel mapping (Mono/Stereo) if len(audio_data.shape) == 1: audio_data = np.vstack([audio_data, audio_data]) else: audio_data = audio_data.T # Shape: (channels, samples) # Trim source offset & duration src_len = audio_data.shape[1] if offset_sample < src_len: actual_dur = min(dur_samples, src_len - offset_sample) sliced_audio = audio_data[:, offset_sample : offset_sample + actual_dur] # Apply gain gain_val = source_data.get("gain", 1.0) sliced_audio = sliced_audio * gain_val # Write to track buffer with boundaries write_end = min(start_sample + sliced_audio.shape[1], total_samples) actual_len = write_end - start_sample if actual_len > 0: track_buffer[:, start_sample:write_end] += sliced_audio[:, :actual_len] except Exception as e: print(f"[RenderEngine] Error reading audio file {resolved_path}: {e}") elif item_type == "MIDI_ITEM": source_data = item.get("source_data", {}) notes = source_data.get("notes", []) # Convert to midi events required by vst_engine midi_events = [] for note in notes: note_start_bar = note["start_beat"] / time_sig_num # Filter notes within the non-destructive visible window offset_bar = item["clip_start_offset_bars"] dur_bar = item["duration_bars"] if note_start_bar >= offset_bar and note_start_bar < (offset_bar + dur_bar): rel_bar_in_item = note_start_bar - offset_bar target_global_bar = item["start_bar"] + rel_bar_in_item midi_events.append({ "note": note["pitch"], "start_beat": target_global_bar * time_sig_num, "duration_beats": note["duration_beats"], "velocity": int(note.get("velocity", 0.8) * 127) }) if midi_events: try: instrument_id = track.get("instrument", "") plugin_mgr = PluginManager() vst = plugin_mgr.load_vst(instrument_id) if instrument_id else None instrument_source = track.get("instrument_source", "soundfont") if instrument_source == "pianobook": dspreset_path = track.get("dspreset_path", "") if dspreset_path and os.path.exists(dspreset_path) and HAS_PEDALBOARD: from pedalboard import Pedalboard ds_manager = DecentSamplerManager() try: vst = ds_manager.create_decent_sampler_instance(dspreset_path) midi_messages = PluginManager.midi_events_to_messages( midi_events, bpm, self.sample_rate ) total_needed = 0 for ev in midi_events: end_sec = (ev.get("start_beat", 0) + ev.get("duration_beats", 1)) * (60.0 / bpm) dur_samples = int(end_sec * self.sample_rate) if dur_samples > total_needed: total_needed = dur_samples total_needed = max(total_needed, 1024) silent = np.zeros((2, total_needed), dtype=np.float32) board = Pedalboard([vst]) synth_buffer = board(silent, sample_rate=self.sample_rate, midi_messages=midi_messages) except Exception as e: print(f"[RenderEngine] DecentSampler/Pianobook error: {e}") synth_buffer = render_midi_events_to_audio( midi_events=midi_events, sr=self.sample_rate, bpm=bpm, instrument='synth' ) else: synth_buffer = render_midi_events_to_audio( midi_events=midi_events, sr=self.sample_rate, bpm=bpm, instrument='synth' ) elif vst and HAS_PEDALBOARD: from pedalboard import Pedalboard # Convert MIDI events with precise sample offset midi_messages = PluginManager.midi_events_to_messages( midi_events, bpm, self.sample_rate, bank=soundfont_bank, program=soundfont_program ) total_needed = 0 for ev in midi_events: end_sec = (ev.get("start_beat", 0) + ev.get("duration_beats", 1)) * (60.0 / bpm) dur_samples = int(end_sec * self.sample_rate) if dur_samples > total_needed: total_needed = dur_samples total_needed = max(total_needed, 1024) silent = np.zeros((2, total_needed), dtype=np.float32) board = Pedalboard([vst]) synth_buffer = board(silent, sample_rate=self.sample_rate, midi_messages=midi_messages) elif instrument_id and instrument_id.startswith("sf_"): sf_path = os.path.join( os.path.dirname(os.path.dirname(__file__)), "static", "soundfonts", instrument_id.replace("sf_", "") + ".sf2" ) if os.path.exists(sf_path) and HAS_PYFLUIDSYNTH: import fluidsynth fl = fluidsynth.FluidSynth(sample_rate=self.sample_rate, gain=0.5) fid = fl.sfload(sf_path) fl.program_select(midi_channel, fid, soundfont_bank, soundfont_program) beat_sec = 60.0 / bpm total_sec = 0 for ev in midi_events: end_sec = (ev.get("start_beat", 0) + ev.get("duration_beats", 1)) * beat_sec if end_sec > total_sec: total_sec = end_sec total_samples = int((total_sec + 1.0) * self.sample_rate) midi_data = np.zeros((2, total_samples), dtype=np.float32) for ev in midi_events: note = ev.get("note", 60) velocity = ev.get("velocity", 100) start_beat = ev.get("start_beat", 0.0) dur_beats = ev.get("duration_beats", 1.0) start_sec = start_beat * beat_sec dur_sec = dur_beats * beat_sec fl.noteon(midi_channel, note, velocity) start_s = int(start_sec * self.sample_rate) dur_s = int(dur_sec * self.sample_rate) block = fl.get_samples(int(dur_s)) if hasattr(fl, 'get_samples') else np.zeros((2, dur_s), dtype=np.float32) fl.noteoff(midi_channel, note) if block.shape[1] > 0: end_s = min(start_s + block.shape[1], total_samples) actual = end_s - start_s if actual > 0: midi_data[:, start_s:end_s] += block[:, :actual] synth_buffer = midi_data fl.delete() else: synth_buffer = render_midi_events_to_audio( midi_events=midi_events, sr=self.sample_rate, bpm=bpm, instrument='synth' ) else: synth_buffer = render_midi_events_to_audio( midi_events=midi_events, sr=self.sample_rate, bpm=bpm, instrument='synth' ) actual_len = min(synth_buffer.shape[1], total_samples) track_buffer[:, :actual_len] += synth_buffer[:, :actual_len] except Exception as e: print(f"[RenderEngine] Error rendering MIDI: {e}") elif item_type == "SECTION_ITEM": source_data = item.get("source_data", {}) sec_id = source_data.get("referenced_section_id", "") if sec_id and sec_id in section_store: # Render nested section recursively sec_container = section_store[sec_id] sec_buffer = self.render_session_container( session=sec_container, section_store=section_store, bpm=bpm, time_sig_num=time_sig_num, total_samples=total_samples ) # Apply non-destructive crop/slicing on section buffer if offset_sample < total_samples: actual_dur = min(dur_samples, total_samples - offset_sample) sliced_sec = sec_buffer[:, offset_sample : offset_sample + actual_dur] # Write to track buffer write_end = min(start_sample + sliced_sec.shape[1], total_samples) actual_len = write_end - start_sample if actual_len > 0: track_buffer[:, start_sample:write_end] += sliced_sec[:, :actual_len] # Apply Track Gain (via Pedalboard or fallback) vol_db = track.get("volume_db", 0.0) pan = track.get("pan", 0.0) mute = track.get("mute", False) if mute: continue # Apply Track FX (Chorus or Reverb) fx_type = track.get("fx_type") if fx_type == "chorus": if HAS_PEDALBOARD: try: board = Pedalboard([Chorus(rate_hz=1.5, depth=0.25)]) track_buffer = board(track_buffer, sample_rate=self.sample_rate) except Exception as e: print(f"[RenderEngine] Pedalboard Chorus failed: {e}") else: # Fallback chorus using simple LFO delay modulation in scipy/numpy try: # 1.5 Hz sine LFO, modulating delay time between 15ms and 25ms (average 20ms) lfo = 0.020 + 0.005 * np.sin(2 * np.pi * 1.5 * np.arange(total_samples) / self.sample_rate) dry = track_buffer * 0.6 wet = np.zeros_like(track_buffer) for ch in range(2): indices = np.arange(total_samples) - (lfo * self.sample_rate) indices = np.clip(indices, 0, total_samples - 1).astype(np.int32) wet[ch, :] = track_buffer[ch, indices] track_buffer = dry + wet * 0.5 except Exception as e: print(f"[RenderEngine] Fallback Chorus failed: {e}") elif fx_type == "reverb": if HAS_PEDALBOARD: try: board = Pedalboard([Reverb(room_size=0.5, wet_level=0.4, dry_level=0.6)]) track_buffer = board(track_buffer, sample_rate=self.sample_rate) except Exception as e: print(f"[RenderEngine] Pedalboard Reverb failed: {e}") else: # Fallback reverb using exponentially decaying noise room impulse response try: # Generate impulse response (decaying noise) len_ir = int(self.sample_rate * 2.0) t_ir = np.arange(len_ir) / self.sample_rate decay = np.exp(-t_ir / 0.5) ir_l = (np.random.rand(len_ir) * 2 - 1) * decay ir_r = (np.random.rand(len_ir) * 2 - 1) * decay dry = track_buffer * 0.6 wet = np.zeros_like(track_buffer) for ch in range(2): ir = ir_l if ch == 0 else ir_r # Convolve conv = signal.convolve(track_buffer[ch, :], ir, mode='full')[:total_samples] wet[ch, :] = conv track_buffer = dry + wet * 0.4 except Exception as e: print(f"[RenderEngine] Fallback Reverb failed: {e}") # Process track volume if HAS_PEDALBOARD: try: board = Pedalboard([Gain(gain_db=vol_db)]) processed_track = board(track_buffer, sample_rate=self.sample_rate) except Exception: gain_linear = 10 ** (vol_db / 20.0) processed_track = track_buffer * gain_linear else: gain_linear = 10 ** (vol_db / 20.0) processed_track = track_buffer * gain_linear # Apply Track Pan if pan != 0.0: # Constant power panning theta = ((np.clip(pan, -1.0, 1.0) + 1.0) / 2.0) * (np.pi / 2.0) processed_track[0, :] *= np.cos(theta) processed_track[1, :] *= np.sin(theta) # Mix track to session session_buffer += processed_track return session_buffer def render_project(self, project_json: dict, output_filepath: str): bpm = project_json["metadata"]["bpm"] time_sig_num = project_json["metadata"].get("time_signature_numerator", 4) main_session = project_json["main_session"] section_store = project_json.get("section_store", {}) # Compute total project samples total_bars = main_session.get("length_bars", 16.0) total_samples = self.bars_to_samples(total_bars, bpm, time_sig_num) # Render main session master_buffer = self.render_session_container( session=main_session, section_store=section_store, bpm=bpm, time_sig_num=time_sig_num, total_samples=total_samples ) # Normalization to prevent clipping max_peak = np.max(np.abs(master_buffer)) if max_peak > 1.0: master_buffer /= max_peak # Write final output file sf.write(output_filepath, master_buffer.T, self.sample_rate) return output_filepath