audio-to-midi-api / processing.py
Basslila's picture
Initial upload of audio-to-midi API
f632d67 verified
Raw
History Blame Contribute Delete
25.2 kB
"""
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