""" 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("= 24: data = f.read(min(chunk_size, 32)) tempo = struct.unpack(" 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