import os import numpy as np import soundfile as sf from app.config import settings from app.core.vst_engine import ( render_midi_events_to_audio, PluginManager, HAS_PEDALBOARD, HAS_PYFLUIDSYNTH, ) if HAS_PEDALBOARD: try: from pedalboard import Pedalboard, Gain 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) for track in session.get("tracks", []): track_type = track.get("type", "AUDIO") track_buffer = np.zeros((2, total_samples), dtype=np.float32) 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 if 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 ) 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(0, fid, 0, 0) 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(0, 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(0, 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 # 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