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 import subprocess import sys def check_pedalboard_safe(): try: res = subprocess.run( [sys.executable, "-c", "import pedalboard"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=2.0 ) return res.returncode == 0 except Exception: return False HAS_PEDALBOARD = check_pedalboard_safe() 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: # Synthesize MIDI track notes synth_buffer = render_midi_events_to_audio( midi_events=midi_events, sr=self.sample_rate, bpm=bpm, instrument='synth' ) # Add to track buffer 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