Spaces:
Running
Running
| """ | |
| Audio Processing Pipeline for Audio-to-MIDI conversion. | |
| Uses Basic Pitch for pitch detection, Librosa for audio analysis, | |
| music21 for chord recognition, and pretty_midi for MIDI generation. | |
| """ | |
| import base64 | |
| import io | |
| import json | |
| import logging | |
| import os | |
| import re | |
| import struct | |
| import tempfile | |
| from typing import Any | |
| import librosa | |
| import music21 | |
| import numpy as np | |
| import pretty_midi | |
| from basic_pitch.inference import predict | |
| try: | |
| import mutagen | |
| from mutagen.id3 import ID3 | |
| from mutagen.flac import FLAC as MutagenFLAC | |
| from mutagen.oggvorbis import OggVorbis | |
| HAS_MUTAGEN = True | |
| except ImportError: | |
| HAS_MUTAGEN = False | |
| logger = logging.getLogger("processing") | |
| logging.basicConfig(level=logging.INFO) | |
| def process_audio(file_path: str) -> dict[str, Any]: | |
| """ | |
| Main processing pipeline: | |
| 1. Run Basic Pitch for pitch detection | |
| 2. Detect BPM and key with Librosa | |
| 3. Analyze chords with music21 | |
| 4. Predict bass notes | |
| 5. Generate chord and bass MIDI files | |
| """ | |
| # --- Step 1: Basic Pitch — Audio to raw MIDI --- | |
| model_output, midi_data, note_events = predict(file_path) | |
| # note_events is a list of (start_time, end_time, pitch_midi, amplitude, pitch_bends) | |
| # amplitude (note[3]) is used as confidence; note[4] is pitch_bends (a list) | |
| if len(note_events) == 0: | |
| return { | |
| "chords": [], | |
| "bass_notes": [], | |
| "key": "Unknown", | |
| "bpm": 0, | |
| "duration": 0, | |
| "confidence": 0, | |
| "chords_midi_base64": None, | |
| "bass_midi_base64": None, | |
| "chromagram_json": "", | |
| } | |
| # --- Step 2: Load audio with Librosa for analysis --- | |
| y, sr = librosa.load(file_path, sr=22050) | |
| duration = librosa.get_duration(y=y, sr=sr) | |
| # BPM detection — priority: 1) file metadata 2) filename hint 3) librosa | |
| metadata_bpm = _extract_bpm_from_metadata(file_path) | |
| filename_bpm = _extract_bpm_from_filename(file_path) | |
| if metadata_bpm: | |
| bpm = metadata_bpm | |
| logger.info(f"BPM from audio metadata: {bpm}") | |
| elif filename_bpm: | |
| bpm = filename_bpm | |
| logger.info(f"BPM from filename: {bpm}") | |
| else: | |
| # Librosa beat tracker (may double the tempo for half-time feels) | |
| tempo, _ = librosa.beat.beat_track(y=y, sr=sr) | |
| if isinstance(tempo, np.ndarray): | |
| raw_bpm = float(tempo[0]) if len(tempo) > 0 else 120.0 | |
| else: | |
| raw_bpm = float(tempo) if tempo else 120.0 | |
| # Half-tempo heuristic: librosa often returns 2x for slow tracks | |
| # If raw > 140 and half is in a musical range (55-100), prefer half | |
| bpm = _apply_half_tempo_heuristic(raw_bpm) | |
| logger.info(f"BPM from librosa: raw={raw_bpm:.1f}, adjusted={bpm:.1f}") | |
| # Key detection using chroma features | |
| chroma = librosa.feature.chroma_cqt(y=y, sr=sr) | |
| detected_key = _detect_key(chroma) | |
| # Beat-aligned chromagram for AI endpoints (much better than 250ms windows) | |
| beat_chromagram = _extract_beat_chromagram(chroma, sr, bpm, duration) | |
| # Calculate exact bar-aligned loop duration | |
| loop_duration = _calculate_loop_length(duration, bpm) | |
| # --- Step 3: Smart filtering — remove low-confidence notes --- | |
| filtered_notes = _filter_notes(note_events, min_confidence=0.4) | |
| # --- Step 4: Chord analysis (use loop_duration for precise boundaries) --- | |
| chords = _analyze_chords(filtered_notes, loop_duration) | |
| # --- Step 5: Bass note prediction --- | |
| bass_notes = _predict_bass(chords, detected_key, bpm) | |
| # --- Step 6: Generate MIDI files (clamped to exact loop length) --- | |
| chords_midi_bytes = _generate_chords_midi(filtered_notes, bpm, loop_duration) | |
| bass_midi_bytes = _generate_bass_midi(bass_notes, bpm, loop_duration) | |
| # Overall confidence (amplitude is at index 3) | |
| if len(filtered_notes) > 0: | |
| avg_confidence = float( | |
| np.mean([n[3] for n in filtered_notes]) | |
| ) | |
| else: | |
| avg_confidence = 0.0 | |
| return { | |
| "chords": chords, | |
| "bass_notes": bass_notes, | |
| "key": detected_key, | |
| "bpm": round(bpm, 1), | |
| "duration": round(duration, 2), | |
| "loop_duration": round(loop_duration, 4), | |
| "confidence": round(avg_confidence, 2), | |
| "chords_midi_base64": base64.b64encode(chords_midi_bytes).decode( | |
| "utf-8" | |
| ), | |
| "bass_midi_base64": base64.b64encode(bass_midi_bytes).decode( | |
| "utf-8" | |
| ), | |
| "chromagram_json": json.dumps(beat_chromagram), | |
| } | |
| # --- BPM Extraction from Metadata --- | |
| def _extract_bpm_from_metadata(file_path: str) -> float | None: | |
| """ | |
| Try to read BPM/tempo from the audio file's metadata. | |
| Supports: | |
| - WAV: ACID chunk (Ableton, FL Studio, Sony ACID exports) | |
| - MP3: ID3 TBPM tag | |
| - FLAC: Vorbis comment BPM/TEMPO | |
| - OGG: Vorbis comment BPM/TEMPO | |
| Returns BPM as float, or None if not found. | |
| """ | |
| ext = os.path.splitext(file_path)[1].lower() | |
| # --- WAV: Parse ACID chunk for tempo --- | |
| if ext == ".wav": | |
| bpm = _extract_bpm_from_wav_acid(file_path) | |
| if bpm: | |
| return bpm | |
| # --- Use mutagen for tag-based formats --- | |
| if not HAS_MUTAGEN: | |
| return None | |
| try: | |
| if ext == ".mp3": | |
| tags = ID3(file_path) | |
| # TBPM is the standard ID3 BPM tag | |
| tbpm = tags.get("TBPM") | |
| if tbpm and tbpm.text: | |
| val = float(tbpm.text[0]) | |
| if 20 < val < 300: | |
| return val | |
| elif ext == ".flac": | |
| audio = MutagenFLAC(file_path) | |
| for key in ("bpm", "BPM", "tempo", "TEMPO"): | |
| vals = audio.get(key) | |
| if vals: | |
| val = float(vals[0]) | |
| if 20 < val < 300: | |
| return val | |
| elif ext == ".ogg": | |
| audio = OggVorbis(file_path) | |
| for key in ("bpm", "BPM", "tempo", "TEMPO"): | |
| vals = audio.get(key) | |
| if vals: | |
| val = float(vals[0]) | |
| if 20 < val < 300: | |
| return val | |
| # Generic mutagen fallback for any format | |
| audio = mutagen.File(file_path, easy=True) | |
| if audio: | |
| for key in ("bpm", "BPM", "tempo", "TEMPO"): | |
| vals = audio.get(key) | |
| if vals: | |
| val = float(vals[0]) | |
| if 20 < val < 300: | |
| return val | |
| except Exception as e: | |
| logger.debug(f"Mutagen metadata read failed: {e}") | |
| return None | |
| def _extract_bpm_from_wav_acid(file_path: str) -> float | None: | |
| """ | |
| Parse WAV RIFF chunks looking for the ACID chunk that stores tempo. | |
| The ACID chunk is used by Ableton, FL Studio, ACID, and many sample packs. | |
| Format: chunk ID 'acid', 24 bytes of data, tempo at offset 12 as float32. | |
| """ | |
| try: | |
| with open(file_path, "rb") as f: | |
| # Verify RIFF header | |
| riff = f.read(4) | |
| if riff != b"RIFF": | |
| return None | |
| f.read(4) # file size | |
| wave = f.read(4) | |
| if wave != b"WAVE": | |
| return None | |
| # Walk through chunks | |
| while True: | |
| chunk_header = f.read(8) | |
| if len(chunk_header) < 8: | |
| break | |
| chunk_id = chunk_header[:4] | |
| chunk_size = struct.unpack("<I", chunk_header[4:8])[0] | |
| if chunk_id == b"acid": | |
| # ACID chunk found! Tempo is at offset 12 (float32 LE) | |
| if chunk_size >= 24: | |
| data = f.read(min(chunk_size, 32)) | |
| tempo = struct.unpack("<f", data[12:16])[0] | |
| if 20 < tempo < 300: | |
| logger.info(f"Found ACID chunk tempo: {tempo}") | |
| return float(tempo) | |
| break | |
| else: | |
| # Skip this chunk (pad to even boundary) | |
| skip = chunk_size + (chunk_size % 2) | |
| f.seek(skip, 1) | |
| except Exception as e: | |
| logger.debug(f"WAV ACID chunk parse failed: {e}") | |
| return None | |
| def _extract_bpm_from_filename(file_path: str) -> float | None: | |
| """ | |
| Look for BPM hints in the filename. | |
| Common patterns: '85bpm', '85_bpm', '85 BPM', 'BPM85', 'tempo85' | |
| """ | |
| basename = os.path.basename(file_path) | |
| name = os.path.splitext(basename)[0] | |
| # Pattern: number followed by 'bpm' (e.g., '85bpm', '85_bpm', '85 bpm') | |
| match = re.search(r'(\d{2,3})\s*[-_]?\s*bpm', name, re.IGNORECASE) | |
| if match: | |
| val = float(match.group(1)) | |
| if 20 < val < 300: | |
| return val | |
| # Pattern: 'bpm' followed by number (e.g., 'bpm85', 'bpm_85') | |
| match = re.search(r'bpm\s*[-_]?\s*(\d{2,3})', name, re.IGNORECASE) | |
| if match: | |
| val = float(match.group(1)) | |
| if 20 < val < 300: | |
| return val | |
| # Pattern: 'tempo' followed by number | |
| match = re.search(r'tempo\s*[-_]?\s*(\d{2,3})', name, re.IGNORECASE) | |
| if match: | |
| val = float(match.group(1)) | |
| if 20 < val < 300: | |
| return val | |
| return None | |
| def _apply_half_tempo_heuristic(raw_bpm: float) -> float: | |
| """ | |
| Librosa's beat tracker often doubles the BPM for half-time feels | |
| (e.g., 85 BPM hip-hop → detected as 170 BPM). | |
| Heuristic: if raw BPM > 140 and halving it gives a value in a | |
| common musical range (55-100), prefer the half value. | |
| This covers hip-hop (70-100), trap (60-90), R&B (60-80), reggaeton (80-100). | |
| """ | |
| if raw_bpm > 140: | |
| half = raw_bpm / 2.0 | |
| if 55 <= half <= 100: | |
| logger.info( | |
| f"Half-tempo heuristic: {raw_bpm:.1f} -> {half:.1f} BPM" | |
| ) | |
| return round(half, 1) | |
| return round(raw_bpm, 1) | |
| # --- Loop Length Calculation --- | |
| def _calculate_loop_length(duration: float, bpm: float) -> float: | |
| """Calculate exact bar-aligned loop duration from audio length and BPM. | |
| Musical loops are always an exact number of bars. Given the audio duration | |
| and BPM, round to the nearest whole number of bars and return the | |
| precise duration in seconds (assuming 4/4 time). | |
| """ | |
| if bpm <= 0: | |
| return duration | |
| beat_duration = 60.0 / bpm | |
| bar_duration = beat_duration * 4 # 4/4 time | |
| total_bars = round(duration / bar_duration) | |
| if total_bars < 1: | |
| total_bars = 1 | |
| loop_duration = total_bars * bar_duration | |
| logger.info( | |
| f"Loop length: {duration:.3f}s audio -> {total_bars} bars " | |
| f"@ {bpm:.1f} BPM = {loop_duration:.4f}s" | |
| ) | |
| return loop_duration | |
| # --- Key Detection --- | |
| KEY_PROFILES = { | |
| "C": [6.35, 2.23, 3.48, 2.33, 4.38, 4.09, 2.52, 5.19, 2.39, 3.66, 2.29, 2.88], | |
| "C#": None, # rotated from C | |
| "D": None, | |
| "D#": None, | |
| "E": None, | |
| "F": None, | |
| "F#": None, | |
| "G": None, | |
| "G#": None, | |
| "A": None, | |
| "A#": None, | |
| "B": None, | |
| } | |
| MAJOR_PROFILE = [6.35, 2.23, 3.48, 2.33, 4.38, 4.09, 2.52, 5.19, 2.39, 3.66, 2.29, 2.88] | |
| MINOR_PROFILE = [6.33, 2.68, 3.52, 5.38, 2.60, 3.53, 2.54, 4.75, 3.98, 2.69, 3.34, 3.17] | |
| NOTE_NAMES = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"] | |
| def _detect_key(chroma: np.ndarray) -> str: | |
| """Detect musical key using Krumhansl-Schmuckler algorithm.""" | |
| chroma_avg = np.mean(chroma, axis=1) | |
| best_corr = -2 | |
| best_key = "C" | |
| best_mode = "major" | |
| for i in range(12): | |
| # Major | |
| major_rotated = np.roll(MAJOR_PROFILE, i) | |
| corr = float(np.corrcoef(chroma_avg, major_rotated)[0, 1]) | |
| if corr > best_corr: | |
| best_corr = corr | |
| best_key = NOTE_NAMES[i] | |
| best_mode = "major" | |
| # Minor | |
| minor_rotated = np.roll(MINOR_PROFILE, i) | |
| corr = float(np.corrcoef(chroma_avg, minor_rotated)[0, 1]) | |
| if corr > best_corr: | |
| best_corr = corr | |
| best_key = NOTE_NAMES[i] | |
| best_mode = "minor" | |
| return f"{best_key} {best_mode}" | |
| # --- Beat-Aligned Chromagram Extraction --- | |
| def _extract_beat_chromagram( | |
| chroma: np.ndarray, sr: int, bpm: float, duration: float | |
| ) -> list[dict]: | |
| """Extract beat-aligned chromagram: pitch class energy at each beat position. | |
| Instead of fixed 250ms windows, aligns to musical beats for more accurate | |
| harmonic analysis. GPT uses these per-beat pitch histograms to identify | |
| the actual chord progression from the audio spectral content. | |
| Args: | |
| chroma: Pre-computed chromagram from librosa.feature.chroma_cqt | |
| sr: Sample rate used for chroma computation | |
| bpm: Detected BPM | |
| duration: Audio duration in seconds | |
| Returns: | |
| List of dicts with beat number, timing, and 12 pitch class energies. | |
| """ | |
| beat_duration = 60.0 / max(bpm, 40) | |
| beat_times = np.arange(0, duration, beat_duration) | |
| if len(beat_times) < 2: | |
| return [] | |
| times = librosa.times_like(chroma, sr=sr, hop_length=512) | |
| result = [] | |
| for i in range(len(beat_times)): | |
| start_t = float(beat_times[i]) | |
| end_t = float(beat_times[i + 1]) if i + 1 < len(beat_times) else duration | |
| # Find chroma frames within this beat | |
| mask = (times >= start_t) & (times < end_t) | |
| if not np.any(mask): | |
| continue | |
| # Average chromagram over this beat | |
| avg = np.mean(chroma[:, mask], axis=1) | |
| # Normalize to 0-1 range | |
| mx = float(np.max(avg)) | |
| if mx > 0: | |
| avg = avg / mx | |
| result.append({ | |
| "beat": i + 1, | |
| "time": round(start_t, 3), | |
| "end_time": round(end_t, 3), | |
| "pitches": [round(float(v), 3) for v in avg], | |
| }) | |
| logger.info(f"Extracted beat chromagram: {len(result)} beats @ {bpm:.0f} BPM") | |
| return result | |
| # --- Note Filtering --- | |
| def _filter_notes( | |
| note_events: list, min_confidence: float = 0.4 | |
| ) -> list: | |
| """Remove ghost notes and low-confidence detections.""" | |
| filtered = [] | |
| for note in note_events: | |
| start_time = note[0] | |
| end_time = note[1] | |
| pitch = note[2] | |
| amplitude = note[3] # amplitude acts as confidence (0.0 - 1.0) | |
| # Filter by amplitude/confidence | |
| if amplitude < min_confidence: | |
| continue | |
| # Filter very short notes (likely artifacts) — less than 50ms | |
| if end_time - start_time < 0.05: | |
| continue | |
| # Filter extremely low or high pitches (likely noise) | |
| if pitch < 24 or pitch > 108: | |
| continue | |
| filtered.append(note) | |
| return filtered | |
| # --- Chord Analysis --- | |
| def _midi_to_note_name(midi_num: int) -> str: | |
| """Convert MIDI number to note name (e.g., 60 -> 'C4').""" | |
| note = NOTE_NAMES[int(midi_num) % 12] | |
| octave = int(midi_num) // 12 - 1 | |
| return f"{note}{octave}" | |
| def _analyze_chords( | |
| note_events: list, duration: float, time_window: float = 0.25 | |
| ) -> list[dict]: | |
| """Group simultaneous notes into chords.""" | |
| if not note_events: | |
| return [] | |
| chords = [] | |
| current_time = 0.0 | |
| while current_time < duration: | |
| window_end = current_time + time_window | |
| # Find notes active in this window | |
| active_notes = [] | |
| for note in note_events: | |
| start, end, pitch = note[0], note[1], int(note[2]) | |
| # Note overlaps with window | |
| if start < window_end and end > current_time: | |
| active_notes.append(pitch) | |
| if active_notes: | |
| # Remove duplicates, sort | |
| unique_pitches = sorted(set(active_notes)) | |
| # Get pitch classes (0-11) | |
| pitch_classes = sorted(set([p % 12 for p in unique_pitches])) | |
| chord_name = _identify_chord(pitch_classes) | |
| note_names = [_midi_to_note_name(p) for p in unique_pitches] | |
| chords.append( | |
| { | |
| "name": chord_name, | |
| "startTime": round(current_time, 3), | |
| "endTime": round(window_end, 3), | |
| "notes": note_names, | |
| "confidence": 0.8, | |
| } | |
| ) | |
| current_time = window_end | |
| # Merge consecutive identical chords | |
| merged = _merge_consecutive_chords(chords) | |
| return merged | |
| def _identify_chord(pitch_classes: list[int]) -> str: | |
| """Identify chord name from pitch classes using interval analysis.""" | |
| if not pitch_classes: | |
| return "N/C" | |
| if len(pitch_classes) == 1: | |
| return NOTE_NAMES[pitch_classes[0]] | |
| # Try each pitch class as root | |
| best_match = None | |
| best_score = 0 | |
| chord_templates = { | |
| "": {0, 4, 7}, # Major | |
| "m": {0, 3, 7}, # Minor | |
| "dim": {0, 3, 6}, # Diminished | |
| "aug": {0, 4, 8}, # Augmented | |
| "7": {0, 4, 7, 10}, # Dominant 7th | |
| "maj7": {0, 4, 7, 11}, # Major 7th | |
| "m7": {0, 3, 7, 10}, # Minor 7th | |
| "sus4": {0, 5, 7}, # Suspended 4th | |
| "sus2": {0, 2, 7}, # Suspended 2nd | |
| } | |
| pc_set = set(pitch_classes) | |
| for root in pitch_classes: | |
| intervals = set([(pc - root) % 12 for pc in pc_set]) | |
| for suffix, template in chord_templates.items(): | |
| # How many template notes are present | |
| matches = len(intervals & template) | |
| score = matches / len(template) | |
| if score > best_score: | |
| best_score = score | |
| best_match = f"{NOTE_NAMES[root]}{suffix}" | |
| return best_match or NOTE_NAMES[pitch_classes[0]] | |
| def _merge_consecutive_chords(chords: list[dict]) -> list[dict]: | |
| """Merge consecutive chords with the same name.""" | |
| if not chords: | |
| return [] | |
| merged = [chords[0].copy()] | |
| for chord in chords[1:]: | |
| if chord["name"] == merged[-1]["name"]: | |
| merged[-1]["endTime"] = chord["endTime"] | |
| # Combine unique notes | |
| all_notes = list( | |
| set(merged[-1]["notes"] + chord["notes"]) | |
| ) | |
| merged[-1]["notes"] = sorted(all_notes) | |
| else: | |
| merged.append(chord.copy()) | |
| return merged | |
| # --- Bass Note Prediction --- | |
| def _predict_bass( | |
| chords: list[dict], key: str, bpm: float | |
| ) -> list[dict]: | |
| """ | |
| Predict bass notes for each chord. | |
| Strategy: | |
| - Use chord root as primary bass note | |
| - Add 5th for alternating bass patterns | |
| - Adjust velocity and pattern based on BPM/genre hints | |
| """ | |
| if not chords: | |
| return [] | |
| bass_notes = [] | |
| for chord in chords: | |
| chord_name = chord["name"] | |
| start = chord["startTime"] | |
| end = chord["endTime"] | |
| duration = end - start | |
| # Extract root note from chord name | |
| root = _extract_root(chord_name) | |
| if root is None: | |
| continue | |
| root_midi = _note_name_to_midi(root, octave=2) # Bass range | |
| fifth_midi = root_midi + 7 # Perfect 5th | |
| # Determine velocity based on BPM | |
| if bpm > 140: | |
| # Fast tempo (EDM) — strong root hits | |
| velocity = 110 | |
| elif bpm > 100: | |
| # Medium (Pop/Rock) — moderate | |
| velocity = 95 | |
| else: | |
| # Slow (Hip-hop/R&B) — sub-bass feel | |
| velocity = 100 | |
| if duration > 0.5: | |
| # Longer chord: root on downbeat + fifth halfway | |
| mid = start + duration / 2 | |
| bass_notes.append( | |
| { | |
| "note": _midi_to_note_name(root_midi), | |
| "startTime": round(start, 3), | |
| "endTime": round(mid, 3), | |
| "velocity": velocity, | |
| } | |
| ) | |
| bass_notes.append( | |
| { | |
| "note": _midi_to_note_name(fifth_midi), | |
| "startTime": round(mid, 3), | |
| "endTime": round(end, 3), | |
| "velocity": int(velocity * 0.8), | |
| } | |
| ) | |
| else: | |
| # Short chord: just root | |
| bass_notes.append( | |
| { | |
| "note": _midi_to_note_name(root_midi), | |
| "startTime": round(start, 3), | |
| "endTime": round(end, 3), | |
| "velocity": velocity, | |
| } | |
| ) | |
| return bass_notes | |
| def _extract_root(chord_name: str) -> str | None: | |
| """Extract root note name from chord name (e.g., 'Am7' -> 'A').""" | |
| if not chord_name or chord_name == "N/C": | |
| return None | |
| # Handle sharps/flats | |
| if len(chord_name) >= 2 and chord_name[1] in ("#", "b"): | |
| return chord_name[:2] | |
| return chord_name[0] | |
| def _note_name_to_midi(note: str, octave: int = 4) -> int: | |
| """Convert note name to MIDI number.""" | |
| note_map = { | |
| "C": 0, "C#": 1, "Db": 1, "D": 2, "D#": 3, "Eb": 3, | |
| "E": 4, "F": 5, "F#": 6, "Gb": 6, "G": 7, "G#": 8, | |
| "Ab": 8, "A": 9, "A#": 10, "Bb": 10, "B": 11, | |
| } | |
| midi = note_map.get(note, 0) | |
| return (octave + 1) * 12 + midi | |
| # --- Quantisation helper --- | |
| def _quantize_16th(t: float, bpm: float) -> float: | |
| """Snap a time value (seconds) to the nearest 1/16-note grid position.""" | |
| sixteenth = 60.0 / bpm / 4.0 | |
| return round(t / sixteenth) * sixteenth | |
| # --- MIDI Generation --- | |
| def _generate_chords_midi(note_events: list, bpm: float, loop_duration: float = 0) -> bytes: | |
| """Generate a MIDI file from detected notes, clamped to exact loop length.""" | |
| midi = pretty_midi.PrettyMIDI(initial_tempo=bpm) | |
| instrument = pretty_midi.Instrument( | |
| program=0, name="Detected Chords" | |
| ) | |
| for note in note_events: | |
| start = float(note[0]) | |
| end = float(note[1]) | |
| pitch = int(note[2]) | |
| amplitude = float(note[3]) # 0.0 - 1.0 | |
| velocity = min(int(amplitude * 127), 127) | |
| # Quantize to 1/16 grid | |
| start = _quantize_16th(start, bpm) | |
| end = _quantize_16th(end, bpm) | |
| # Clamp to exact loop bounds | |
| if loop_duration > 0: | |
| start = max(0.0, min(start, loop_duration)) | |
| end = max(0.0, min(end, loop_duration)) | |
| if end <= start: | |
| continue | |
| midi_note = pretty_midi.Note( | |
| velocity=velocity, | |
| pitch=pitch, | |
| start=start, | |
| end=end, | |
| ) | |
| instrument.notes.append(midi_note) | |
| # Force MIDI file to span exactly loop_duration with CC#123 (All Notes Off) | |
| if loop_duration > 0: | |
| instrument.control_changes.append( | |
| pretty_midi.ControlChange(number=123, value=0, time=loop_duration) | |
| ) | |
| midi.instruments.append(instrument) | |
| buffer = io.BytesIO() | |
| midi.write(buffer) | |
| return buffer.getvalue() | |
| def _generate_bass_midi( | |
| bass_notes: list[dict], bpm: float, loop_duration: float = 0 | |
| ) -> bytes: | |
| """Generate a MIDI file from predicted bass notes, clamped to exact loop length.""" | |
| midi = pretty_midi.PrettyMIDI(initial_tempo=bpm) | |
| instrument = pretty_midi.Instrument( | |
| program=33, name="Predicted Bass" | |
| ) # program 33 = Fingered Bass | |
| for note_info in bass_notes: | |
| note_name = note_info["note"] | |
| # Parse note name to MIDI | |
| pitch = _parse_note_to_midi(note_name) | |
| if pitch is None: | |
| continue | |
| start = note_info["startTime"] | |
| end = note_info["endTime"] | |
| # Quantize to 1/16 grid | |
| start = _quantize_16th(start, bpm) | |
| end = _quantize_16th(end, bpm) | |
| # Clamp to exact loop bounds | |
| if loop_duration > 0: | |
| start = max(0.0, min(start, loop_duration)) | |
| end = max(0.0, min(end, loop_duration)) | |
| if end <= start: | |
| continue | |
| midi_note = pretty_midi.Note( | |
| velocity=note_info["velocity"], | |
| pitch=pitch, | |
| start=start, | |
| end=end, | |
| ) | |
| instrument.notes.append(midi_note) | |
| # Force MIDI file to span exactly loop_duration with CC#123 (All Notes Off) | |
| if loop_duration > 0: | |
| instrument.control_changes.append( | |
| pretty_midi.ControlChange(number=123, value=0, time=loop_duration) | |
| ) | |
| midi.instruments.append(instrument) | |
| buffer = io.BytesIO() | |
| midi.write(buffer) | |
| return buffer.getvalue() | |
| def _parse_note_to_midi(note_name: str) -> int | None: | |
| """Parse a note name like 'C2' or 'F#3' to MIDI number.""" | |
| note_map = { | |
| "C": 0, "C#": 1, "Db": 1, "D": 2, "D#": 3, "Eb": 3, | |
| "E": 4, "F": 5, "F#": 6, "Gb": 6, "G": 7, "G#": 8, | |
| "Ab": 8, "A": 9, "A#": 10, "Bb": 10, "B": 11, | |
| } | |
| try: | |
| # Extract note and octave | |
| if len(note_name) >= 3 and note_name[1] in ("#", "b"): | |
| note = note_name[:2] | |
| octave = int(note_name[2:]) | |
| elif len(note_name) >= 2: | |
| note = note_name[0] | |
| octave = int(note_name[1:]) | |
| else: | |
| return None | |
| midi_num = note_map.get(note) | |
| if midi_num is None: | |
| return None | |
| return (octave + 1) * 12 + midi_num | |
| except (ValueError, IndexError): | |
| return None | |