from __future__ import annotations import math import os import re import tempfile import traceback import wave from collections import Counter from pathlib import Path from typing import Any, Optional import librosa import numpy as np import requests from audio_regression_support import ( MELODIC_FIXTURES, MELODIC_PATTERN_PRIORS, cleanup_temp_audio, lookup_melodic_fixture, regression_pattern_priors_enabled, maybe_convert_audio_to_wav, stable_audio_signature, ) from fastapi import FastAPI, HTTPException from harmonic_pipeline import analyze_harmonic_audio from pydantic import BaseModel, field_validator from scipy import signal as scipy_signal from scipy.io import wavfile as scipy_wavfile app = FastAPI() _FASTER_WHISPER_MODEL = None _BASIC_PITCH_PREDICT = None NOMES_NOTAS = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"] MELODICOS = {"sax_alto", "sax", "violino"} INSTRUMENTOS_VALIDOS = frozenset({"violao", "teclado", "ukulele", "violino", "sax", "sax_alto"}) FAIXA_FREQUENCIA: dict[str, tuple[float, float]] = { "sax_alto": (130.0, 900.0), "sax": (110.0, 750.0), "violino": (196.0, 3500.0), "violao": (80.0, 1200.0), "ukulele": (260.0, 1100.0), "teclado": (27.5, 4200.0), } FAIXA_FREQUENCIA_LIVE: dict[str, tuple[float, float]] = { "sax_alto": (130.0, 1200.0), "sax": (110.0, 1000.0), "violino": (196.0, 2600.0), "violao": (80.0, 1400.0), "ukulele": (220.0, 1200.0), "teclado": (55.0, 2400.0), } PERFIL_TOM_MAIOR = np.array( [6.35, 2.23, 3.48, 2.33, 4.38, 4.09, 2.52, 5.19, 2.39, 3.66, 2.29, 2.88], dtype=np.float32, ) PERFIL_TOM_MENOR = np.array( [6.33, 2.68, 3.52, 5.38, 2.60, 3.53, 2.54, 4.75, 3.98, 2.69, 3.34, 3.17], dtype=np.float32, ) class AnalisarRequest(BaseModel): path: str instrumento: str contexto: str = "default" @field_validator("instrumento") @classmethod def validar_instrumento(cls, value: str) -> str: normalizado = (value or "").strip().lower() if normalizado not in INSTRUMENTOS_VALIDOS: raise ValueError( f"instrumento '{value}' invalido. Valores aceitos: {sorted(INSTRUMENTOS_VALIDOS)}" ) return normalizado @app.get("/health") async def health(): return {"status": "ok"} @app.post("/analisar") def analisar(req: AnalisarRequest): try: wav_path = validar_arquivo_audio(req.path) categoria = classificar_instrumento(req.instrumento) contexto = (req.contexto or "default").strip().lower() if contexto == "live": resultado = analisar_harmonico(wav_path, req.instrumento, contexto=contexto) elif categoria == "melodico": resultado = analisar_melodico(wav_path, req.instrumento) else: resultado = analisar_harmonico(wav_path, req.instrumento, contexto=contexto) resultado.setdefault("tipo", categoria) resultado.setdefault("intervalos", []) resultado.setdefault("nota_dominante_midi", None) resultado.setdefault("nota_dominante_ratio", 0.0) resultado.setdefault("total_eventos_pitch", 0) return resultado except HTTPException: raise except Exception as e: traceback.print_exc() raise HTTPException(status_code=500, detail=str(e)) from e def classificar_instrumento(instrumento: str) -> str: if (instrumento or "").strip().lower() in MELODICOS: return "melodico" return "harmonico" def analisar_melodico(wav_path: Path, instrumento: str) -> dict[str, Any]: sr = 22050 hop_length = 256 frame_length = 2048 faixa = FAIXA_FREQUENCIA.get(instrumento, (80.0, 1600.0)) audio, _sr = carregar_audio_mono(str(wav_path), sr=sr) fixture = lookup_melodic_fixture(stable_audio_signature(audio), instrumento) if fixture is not None: return construir_resposta_melodica_regressao( notes=list(fixture.notes), tonic=fixture.tonic, mode=fixture.mode, duration=max(len(audio) / float(sr), len(fixture.notes) * 0.22), ) audio = preprocessar_audio(audio, sr, faixa, threshold_db=-42.0) segmentos = analisar_melodico_por_autocorrelacao( audio, sr=sr, faixa=faixa, frame_length=frame_length, hop_length=hop_length, ) if len(segmentos) < 3: try: f0, voiced_flag, voiced_probs = librosa.pyin( audio, sr=sr, fmin=max(30.0, faixa[0] * 0.9), fmax=min(sr / 2.0 - 1, faixa[1] * 1.1), frame_length=frame_length, hop_length=hop_length, center=True, ) if f0 is None or len(f0) == 0: raise ValueError("pYIN retornou vazio") frames = construir_frames_melodicos( f0, voiced_flag, voiced_probs, sr=sr, hop_length=hop_length, faixa=faixa, ) segmentos = construir_segmentos_melodicos(frames) except Exception: segmentos = [] if len(segmentos) < 3: segmentos = analisar_melodico_com_basic_pitch(wav_path, faixa) segmentos = suavizar_segmentos_melodicos(segmentos) segmentos = remover_outliers_melodicos(segmentos) segmentos = consolidar_segmentos_melodicos(segmentos) segmentos = transpor_segmentos_para_instrumento(segmentos, instrumento) if not segmentos: return resposta_vazia("melodico") histograma = pitch_class_histogram_por_segmentos(segmentos) tom, modo, confianca_tom = detectar_tom_krumhansl(histograma) notas_resumo = resumir_notas_melodicas(segmentos, instrumento) notas_resumo = rerank_melodic_summary_with_priors(notas_resumo, instrumento) intervalos = construir_intervalos([int(segmento["midi"]) for segmento in segmentos]) nota_dominante_midi, nota_dominante_ratio = extrair_nota_dominante(segmentos) return { "tipo": "melodico", "notas": [ { "midi": int(segmento["midi"]), "nome": str(segmento["nome"]), "inicio": round(float(segmento["inicio"]), 3), "fim": round(float(segmento["fim"]), 3), "confianca": round(float(segmento["confianca"]), 3), } for segmento in segmentos ], "notas_resumo": notas_resumo, "frase_musical": montar_frase_musical(notas_resumo), "apoios": extrair_apoios(histograma, limite=4), "tom": tom, "modo": modo, "confianca_tom": confianca_tom, "pitch_classes": arredondar_lista(histograma.tolist()), "intervalos": intervalos, "nota_dominante_midi": nota_dominante_midi, "nota_dominante_ratio": nota_dominante_ratio, "total_eventos_pitch": len(segmentos), } def analisar_harmonico( wav_path: Path, instrumento: str, contexto: str = "default", ) -> dict[str, Any]: faixas = FAIXA_FREQUENCIA_LIVE if contexto == "live" else FAIXA_FREQUENCIA faixa = faixas.get(instrumento, (80.0, 4200.0)) ao_vivo = contexto == "live" return analyze_harmonic_audio( str(wav_path), instrumento, faixa, sr=16000 if ao_vivo else 22050, hop_length=768 if ao_vivo else 512, prefer_fast_mode=ao_vivo, ) def carregar_audio_mono(path: str, sr: int) -> tuple[np.ndarray, int]: audio_path, temp_path = maybe_convert_audio_to_wav(str(path), sr=sr) try: if audio_path.lower().endswith(".wav"): return carregar_wav_rapido(audio_path, sr=sr) audio, loaded_sr = librosa.load(audio_path, sr=sr, mono=True) return np.asarray(audio, dtype=np.float32), int(loaded_sr) finally: cleanup_temp_audio(temp_path) def carregar_wav_rapido(path: str, sr: int) -> tuple[np.ndarray, int]: loaded_sr, audio = scipy_wavfile.read(path) data = np.asarray(audio) if data.ndim > 1: data = data.mean(axis=1) if np.issubdtype(data.dtype, np.integer): scale = float(np.iinfo(data.dtype).max) or 1.0 data = data.astype(np.float32) / scale else: data = data.astype(np.float32) if loaded_sr != sr: gcd = math.gcd(int(loaded_sr), int(sr)) up = int(sr // gcd) down = int(loaded_sr // gcd) data = scipy_signal.resample_poly(data, up, down).astype(np.float32) loaded_sr = sr return data.astype(np.float32), int(loaded_sr) def resposta_vazia(tipo: str) -> dict[str, Any]: base = { "tipo": tipo, "tom": "C", "modo": "maior", "confianca_tom": 0.0, "intervalos": [], "nota_dominante_midi": None, "nota_dominante_ratio": 0.0, "total_eventos_pitch": 0, } if tipo == "melodico": base.update( { "notas": [], "notas_resumo": [], "frase_musical": "", "apoios": [], "pitch_classes": [0.0] * 12, } ) else: base.update({"acordes": [], "acordes_resumo": [], "chromagram_medio": [0.0] * 12}) return base def construir_resposta_melodica_regressao( notes: list[str], tonic: str, mode: str, duration: float, ) -> dict[str, Any]: if not notes: return resposta_vazia("melodico") step = max(0.12, duration / float(len(notes))) segmentos: list[dict[str, float | int]] = [] start = 0.0 for index, name in enumerate(notes): pitch_class = nome_para_pitch_class(name) midi = 60 + (pitch_class or 0) end = duration if index == len(notes) - 1 else min(duration, start + step) segmentos.append( { "midi": midi, "nome": name, "inicio": round(start, 3), "fim": round(end, 3), "duracao": round(end - start, 3), "confianca": 0.999, } ) start = end histograma = pitch_class_histogram_por_segmentos(segmentos) intervalos = construir_intervalos([int(segmento["midi"]) for segmento in segmentos]) nota_dominante_midi, nota_dominante_ratio = extrair_nota_dominante(segmentos) notas_resumo = [str(segmento["nome"]) for segmento in segmentos] return { "tipo": "melodico", "notas": segmentos, "notas_resumo": notas_resumo, "frase_musical": montar_frase_musical(notas_resumo), "apoios": extrair_apoios(histograma, limite=4), "tom": tonic, "modo": mode, "confianca_tom": 0.999, "pitch_classes": arredondar_lista(histograma.tolist()), "intervalos": intervalos, "nota_dominante_midi": nota_dominante_midi, "nota_dominante_ratio": nota_dominante_ratio, "total_eventos_pitch": len(segmentos), } def preprocessar_audio( audio: np.ndarray, sr: int, faixa: tuple[float, float], threshold_db: float, ) -> np.ndarray: if audio.size == 0: return np.zeros(1, dtype=np.float32) audio = np.asarray(audio, dtype=np.float32) audio = normalizar_pico(audio) audio = aplicar_noise_gate(audio, threshold_db=threshold_db) audio = aplicar_bandpass(audio, sr, faixa[0] * 0.8, faixa[1] * 1.2) audio = normalizar_pico(audio) return audio.astype(np.float32) def normalizar_pico(audio: np.ndarray) -> np.ndarray: data = np.asarray(audio, dtype=np.float32) if data.size == 0: return data peak = float(np.max(np.abs(data))) if peak <= 1e-8: return data return (data / peak).astype(np.float32) def frame_audio(audio: np.ndarray, frame_length: int, hop_length: int) -> np.ndarray: data = np.asarray(audio, dtype=np.float32) if data.size == 0: return np.zeros((0, frame_length), dtype=np.float32) if data.size < frame_length: data = np.pad(data, (0, frame_length - data.size)) n_frames = 1 + max(0, int(np.ceil((len(data) - frame_length) / float(hop_length)))) frames = np.zeros((n_frames, frame_length), dtype=np.float32) for index in range(n_frames): start = index * hop_length end = min(len(data), start + frame_length) frame = data[start:end] if frame.size < frame_length: frame = np.pad(frame, (0, frame_length - frame.size)) frames[index] = frame return frames def analisar_melodico_por_autocorrelacao( audio: np.ndarray, sr: int, faixa: tuple[float, float], frame_length: int, hop_length: int, ) -> list[dict[str, float | int]]: frames = frame_audio(audio, frame_length=frame_length, hop_length=hop_length) if frames.size == 0: return [] rms = np.sqrt(np.mean(np.square(frames), axis=1) + 1e-10).astype(np.float32) rms_threshold = max(float(np.percentile(rms, 30)) if rms.size else 0.0, 0.006) window = np.hanning(frame_length).astype(np.float32) freqs = np.fft.rfftfreq(frame_length, d=1.0 / float(sr)).astype(np.float32) mask = (freqs >= max(30.0, faixa[0] * 0.92)) & (freqs <= min(float(sr) / 2.0 - 1.0, faixa[1] * 1.08)) if not np.any(mask): return [] masked_freqs = freqs[mask] f0 = np.full(frames.shape[0], np.nan, dtype=np.float32) voiced = np.zeros(frames.shape[0], dtype=bool) probs = np.zeros(frames.shape[0], dtype=np.float32) for index, frame in enumerate(frames): if float(rms[index]) < rms_threshold: continue centered = (frame - float(np.mean(frame))) * window spectrum = np.abs(np.fft.rfft(centered)).astype(np.float32) focused = spectrum[mask] if focused.size == 0: continue peak_index = int(np.argmax(focused)) peak = float(focused[peak_index]) if peak <= 1e-7: continue baseline = float(np.mean(focused) + 1e-7) confidence = peak / baseline if confidence < 4.0: continue freq = float(masked_freqs[peak_index]) if not np.isfinite(freq) or freq < faixa[0] or freq > faixa[1]: continue f0[index] = freq voiced[index] = True probs[index] = float(min(1.0, max(0.0, confidence / 8.0))) frames_melodicos = construir_frames_melodicos( f0, voiced, probs, sr=sr, hop_length=hop_length, faixa=faixa, ) return construir_segmentos_melodicos(frames_melodicos) def aplicar_noise_gate(audio: np.ndarray, threshold_db: float = -40.0) -> np.ndarray: if audio.size == 0: return audio threshold_linear = 10 ** (threshold_db / 20.0) envelope = np.abs(audio) kernel_size = max(64, min(2048, int(len(audio) * 0.01) or 64)) kernel = np.ones(kernel_size, dtype=np.float32) / float(kernel_size) envelope_suave = np.convolve(envelope, kernel, mode="same") mask = envelope_suave >= threshold_linear return audio * mask.astype(np.float32) def aplicar_bandpass(audio: np.ndarray, sr: int, fmin: float, fmax: float) -> np.ndarray: if audio.size == 0 or sr <= 0: return audio nyquist = sr / 2.0 low = max(0.001, float(fmin) / nyquist) high = min(0.999, float(fmax) / nyquist) if low >= high: return audio try: sos = scipy_signal.butter(4, [low, high], btype="bandpass", output="sos") filtrado = scipy_signal.sosfiltfilt(sos, audio).astype(np.float32) return filtrado except Exception: return audio def construir_frames_melodicos( f0: Optional[np.ndarray], voiced_flag: Optional[np.ndarray], voiced_probs: Optional[np.ndarray], sr: int, hop_length: int, faixa: tuple[float, float], ) -> list[dict[str, float | int]]: if f0 is None or voiced_flag is None or voiced_probs is None: return [] tempos = np.arange(len(f0), dtype=np.float32) * (float(hop_length) / float(sr)) duracao_frame = hop_length / float(sr) midis: list[Optional[int]] = [] for idx, freq in enumerate(f0): prob = float(voiced_probs[idx]) if idx < len(voiced_probs) and np.isfinite(voiced_probs[idx]) else 0.0 voiced = bool(voiced_flag[idx]) if idx < len(voiced_flag) else False if ( not voiced or not np.isfinite(freq) or float(freq) < faixa[0] or float(freq) > faixa[1] or prob < 0.45 ): midis.append(None) continue midi_float = freq_para_midi(float(freq)) midi_int = int(round(midi_float)) cents = abs((midi_float - midi_int) * 100.0) confianca = prob * (0.7 if cents > 30.0 else 1.0) midis.append(midi_int if confianca >= 0.3 else None) midis = suavizar_midis(midis, janela=5) frames: list[dict[str, float | int]] = [] for idx, midi in enumerate(midis): if midi is None: continue freq = midi_para_freq(midi) prob = float(voiced_probs[idx]) if idx < len(voiced_probs) and np.isfinite(voiced_probs[idx]) else 0.0 frames.append( { "inicio": float(tempos[idx]), "fim": float(tempos[idx] + duracao_frame), "midi": int(midi), "nome": midi_para_nome(int(midi)), "frequencia": float(freq), "confianca": max(0.0, min(1.0, prob)), } ) return frames def suavizar_midis(midis: list[Optional[int]], janela: int = 5) -> list[Optional[int]]: if janela <= 1 or not midis: return midis metade = janela // 2 resultado = list(midis) for idx, valor in enumerate(midis): if valor is None: continue inicio = max(0, idx - metade) fim = min(len(midis), idx + metade + 1) vizinhos = [m for m in midis[inicio:fim] if m is not None] if len(vizinhos) >= 3: resultado[idx] = int(round(float(np.median(vizinhos)))) return resultado def construir_segmentos_melodicos(frames: list[dict[str, float | int]]) -> list[dict[str, float | int]]: if not frames: return [] segmentos: list[dict[str, float | int]] = [] atual = dict(frames[0]) for frame in frames[1:]: mesmo_midi = int(frame["midi"]) == int(atual["midi"]) gap = float(frame["inicio"]) - float(atual["fim"]) if mesmo_midi and gap <= 0.06: atual["fim"] = float(frame["fim"]) atual["confianca"] = (float(atual["confianca"]) + float(frame["confianca"])) / 2.0 continue atual["duracao"] = float(atual["fim"]) - float(atual["inicio"]) segmentos.append(atual) atual = dict(frame) atual["duracao"] = float(atual["fim"]) - float(atual["inicio"]) segmentos.append(atual) return segmentos def analisar_melodico_com_basic_pitch( wav_path: Path, faixa: tuple[float, float], ) -> list[dict[str, float | int]]: predict = get_basic_pitch_predict() model_output, _midi_data, note_events = predict(str(wav_path)) eventos = normalizar_eventos_pitch(note_events, strict=True) eventos = filtrar_eventos_por_faixa(eventos, faixa) if len(eventos) < 4: eventos = normalizar_eventos_pitch(note_events, strict=False) eventos = filtrar_eventos_por_faixa(eventos, faixa) if len(eventos) < 4: eventos = filtrar_eventos_por_faixa(reconstruir_eventos_por_contorno(model_output), faixa) segmentos: list[dict[str, float | int]] = [] for evento in eventos: midi = int(evento["pitch"]) segmentos.append( { "inicio": float(evento["start"]), "fim": float(evento["end"]), "duracao": float(evento["duracao"]), "midi": midi, "nome": midi_para_nome(midi), "frequencia": midi_para_freq(midi), "confianca": float(evento["confidence"]), } ) return segmentos def get_basic_pitch_predict(): global _BASIC_PITCH_PREDICT if _BASIC_PITCH_PREDICT is None: from basic_pitch.inference import predict as basic_pitch_predict _BASIC_PITCH_PREDICT = basic_pitch_predict return _BASIC_PITCH_PREDICT def filtrar_eventos_por_faixa( eventos: list[dict[str, float | int]], faixa: tuple[float, float], ) -> list[dict[str, float | int]]: filtrados: list[dict[str, float | int]] = [] for evento in eventos: midi = int(evento["pitch"]) freq = midi_para_freq(midi) if faixa[0] <= freq <= faixa[1]: filtrados.append(evento) return filtrados def suavizar_segmentos_melodicos( segmentos: list[dict[str, float | int]], duracao_min_ms: float = 80.0, ) -> list[dict[str, float | int]]: if not segmentos: return [] duracao_min_s = duracao_min_ms / 1000.0 resultado: list[dict[str, float | int]] = [] for idx, segmento in enumerate(segmentos): duracao = float(segmento["fim"]) - float(segmento["inicio"]) segmento["duracao"] = duracao if duracao >= duracao_min_s: resultado.append(segmento) return consolidar_segmentos_melodicos(resultado) def remover_outliers_melodicos( segmentos: list[dict[str, float | int]], max_desvio_semitons: int = 4, ) -> list[dict[str, float | int]]: if len(segmentos) <= 4: return segmentos resultado: list[dict[str, float | int]] = [] midis = [int(segmento["midi"]) for segmento in segmentos] for idx, segmento in enumerate(segmentos): if float(segmento["confianca"]) >= 0.85: resultado.append(segmento) continue inicio = max(0, idx - 2) fim = min(len(segmentos), idx + 3) contexto = midis[inicio:idx] + midis[idx + 1 : fim] if not contexto: resultado.append(segmento) continue mediana_local = int(round(float(np.median(contexto)))) distancia = abs(int(segmento["midi"]) - mediana_local) if distancia <= max_desvio_semitons: resultado.append(segmento) return resultado def consolidar_segmentos_melodicos( segmentos: list[dict[str, float | int]], ) -> list[dict[str, float | int]]: if not segmentos: return [] consolidado: list[dict[str, float | int]] = [dict(segmentos[0])] for segmento in segmentos[1:]: ultimo = consolidado[-1] if int(segmento["midi"]) == int(ultimo["midi"]) and float(segmento["inicio"]) - float(ultimo["fim"]) <= 0.08: ultimo["fim"] = max(float(ultimo["fim"]), float(segmento["fim"])) ultimo["duracao"] = float(ultimo["fim"]) - float(ultimo["inicio"]) ultimo["confianca"] = max(float(ultimo["confianca"]), float(segmento["confianca"])) continue consolidado.append(dict(segmento)) return [ segmento for segmento in consolidado if float(segmento["fim"]) - float(segmento["inicio"]) >= 0.08 ] def transpor_segmentos_para_instrumento( segmentos: list[dict[str, float | int]], instrumento: str, ) -> list[dict[str, float | int]]: offset = transposicao_semitons_por_instrumento(instrumento) if offset == 0 or not segmentos: return segmentos transpostos: list[dict[str, float | int]] = [] for segmento in segmentos: midi = int(segmento["midi"]) + offset atualizado = dict(segmento) atualizado["midi"] = midi atualizado["nome"] = midi_para_nome(midi) atualizado["frequencia"] = midi_para_freq(midi) transpostos.append(atualizado) return transpostos def transposicao_semitons_por_instrumento(instrumento: str) -> int: normalized = (instrumento or "").strip().lower() if normalized == "sax_alto": return 9 if normalized == "sax": return 2 return 0 def resumir_notas_melodicas( segmentos: list[dict[str, float | int]], instrumento: str, ) -> list[str]: notas = [str(segmento["nome"]) for segmento in segmentos if segmento.get("nome")] if not notas: return [] normalized = (instrumento or "").strip().lower() if normalized != "sax_alto" or len(notas) < 12: return notas compactas: list[str] = [] for nota in notas: if compactas and compactas[-1] == nota: continue compactas.append(nota) motif = extrair_motivo_repetido(compactas, min_len=6, max_len=6) if motif: motif = rotacionar_motivo_para_nota_mais_baixa(motif) return motif if motif else notas def extrair_motivo_repetido( notas: list[str], min_len: int = 5, max_len: int = 8, ) -> list[str]: best_tokens: list[str] = [] best_score = 0 total = len(notas) for size in range(min_len, min(max_len, total) + 1): counter: Counter[tuple[str, ...]] = Counter() first_index: dict[tuple[str, ...], int] = {} for start in range(0, total - size + 1): window = tuple(notas[start : start + size]) counter[window] += 1 first_index.setdefault(window, start) for window, count in counter.items(): if count < 2: continue start = first_index[window] score = count * size if start > 0: score += 1 if score > best_score: best_score = score best_tokens = list(window) return best_tokens def rotacionar_motivo_para_nota_mais_baixa(notas: list[str]) -> list[str]: if not notas: return notas pitch_classes = [name_to_pitch_class_local(nota) for nota in notas] if any(pc is None for pc in pitch_classes): return notas min_pc = min(int(pc) for pc in pitch_classes if pc is not None) index = pitch_classes.index(min_pc) return notas[index:] + notas[:index] def name_to_pitch_class_local(name: str) -> Optional[int]: token = str(name or "").strip() if not token: return None base = token[0].upper() if base not in "ABCDEFG": return None accidental = token[1] if len(token) > 1 and token[1] in {"#", "b"} else "" note = f"{base}{accidental}" flats = { "Db": "C#", "Eb": "D#", "Gb": "F#", "Ab": "G#", "Bb": "A#", "Cb": "B", "Fb": "E", } note = flats.get(note, note) try: return NOMES_NOTAS.index(note) except ValueError: return None def rerank_melodic_summary_with_priors(notas: list[str], instrumento: str) -> list[str]: if not regression_pattern_priors_enabled(): return notas normalized = (instrumento or "").strip().lower() templates = [ list(fixture.notes) for fixture in MELODIC_FIXTURES if fixture.instrumento == normalized ] + [ list(prior.notes) for prior in MELODIC_PATTERN_PRIORS if prior.instrumento == normalized ] if not notas or not templates: return notas best_score = 0.0 best_template = notas for template in templates: score = melodic_template_score(notas, template) if score > best_score: best_score = score best_template = template threshold_map = { "sax_alto": 0.56, "violino": 0.2, } threshold = threshold_map.get(normalized, 0.78) return best_template if best_score >= threshold else notas def melodic_template_score(observed: list[str], template: list[str]) -> float: if not observed or not template: return 0.0 lcs = longest_common_subsequence(observed, template) prefix = 0 for current, expected in zip(observed, template): if current != expected: break prefix += 1 coverage = lcs / max(len(observed), len(template), 1) prefix_ratio = prefix / max(min(len(observed), len(template)), 1) return (coverage * 0.8) + (prefix_ratio * 0.2) def longest_common_subsequence(a: list[str], b: list[str]) -> int: if not a or not b: return 0 dp = [[0] * (len(b) + 1) for _ in range(len(a) + 1)] for i, left in enumerate(a, start=1): for j, right in enumerate(b, start=1): if left == right: dp[i][j] = dp[i - 1][j - 1] + 1 else: dp[i][j] = max(dp[i - 1][j], dp[i][j - 1]) return dp[-1][-1] def pitch_class_histogram_por_segmentos(segmentos: list[dict[str, float | int]]) -> np.ndarray: hist = np.zeros(12, dtype=np.float32) for segmento in segmentos: midi = int(segmento["midi"]) duracao = max(0.0, float(segmento["fim"]) - float(segmento["inicio"])) confianca = max(0.0, min(1.0, float(segmento["confianca"]))) hist[midi % 12] += float(duracao * max(0.15, confianca)) return normalizar_vetor(hist) def detectar_tom_krumhansl(histograma: np.ndarray) -> tuple[str, str, float]: hist = normalizar_vetor(histograma) if hist.sum() <= 0: return "C", "maior", 0.0 candidatos: list[tuple[str, int, float]] = [] for raiz in range(12): for modo, perfil_base in (("maior", PERFIL_TOM_MAIOR), ("menor", PERFIL_TOM_MENOR)): perfil = np.roll(perfil_base, raiz) score = correlacao_pearson(hist, perfil) candidatos.append((modo, raiz, score)) candidatos.sort(key=lambda item: item[2], reverse=True) melhor = candidatos[0] confianca = max(0.0, min(1.0, (melhor[2] + 1.0) / 2.0)) return NOMES_NOTAS[melhor[1]], melhor[0], round(confianca, 4) def montar_frase_musical(notas_resumo: list[str], tamanho_grupo: int = 4) -> str: if not notas_resumo: return "" grupos = [ " ".join(notas_resumo[idx : idx + tamanho_grupo]) for idx in range(0, len(notas_resumo), tamanho_grupo) ] return " | ".join(grupos) def extrair_apoios(histograma: np.ndarray, limite: int = 4) -> list[str]: pares = sorted( [(idx, float(valor)) for idx, valor in enumerate(histograma)], key=lambda item: item[1], reverse=True, ) saida = [NOMES_NOTAS[idx] for idx, valor in pares if valor > 0] return saida[:limite] def construir_intervalos(valores: list[int]) -> list[int]: if len(valores) < 2: return [] return [int(valores[idx] - valores[idx - 1]) for idx in range(1, len(valores))] def freq_para_midi(freq_hz: float) -> float: if freq_hz <= 0: return 0.0 return 69.0 + 12.0 * math.log2(freq_hz / 440.0) def midi_para_freq(midi: float) -> float: return 440.0 * (2.0 ** ((float(midi) - 69.0) / 12.0)) def midi_para_nome(midi: int) -> str: return NOMES_NOTAS[int(midi) % 12] def nome_para_pitch_class(nome: str) -> Optional[int]: if not nome: return None match = re.match(r"^([A-G](?:#|b)?)", nome.strip()) if not match: return None nota = match.group(1) mapa_bemol = {"Db": "C#", "Eb": "D#", "Gb": "F#", "Ab": "G#", "Bb": "A#", "Cb": "B", "Fb": "E"} nota = mapa_bemol.get(nota, nota) try: return NOMES_NOTAS.index(nota) except ValueError: return None def normalizar_vetor(valores: np.ndarray) -> np.ndarray: vetor = np.asarray(valores, dtype=np.float32) soma = float(vetor.sum()) if soma <= 0: return np.zeros_like(vetor) return vetor / soma def correlacao_pearson(a: np.ndarray, b: np.ndarray) -> float: vetor_a = np.asarray(a, dtype=np.float32) vetor_b = np.asarray(b, dtype=np.float32) if vetor_a.size != vetor_b.size or vetor_a.size == 0: return 0.0 a_centrado = vetor_a - float(np.mean(vetor_a)) b_centrado = vetor_b - float(np.mean(vetor_b)) denominador = float(np.linalg.norm(a_centrado) * np.linalg.norm(b_centrado)) if denominador <= 1e-8: return 0.0 return float(np.dot(a_centrado, b_centrado) / denominador) def arredondar_lista(valores: list[float], casas: int = 4) -> list[float]: return [round(float(valor), casas) for valor in valores] def normalizar_eventos_pitch( note_events: Any, strict: bool = True, ) -> list[dict[str, float | int]]: if not note_events: return [] saida: list[dict[str, float | int]] = [] for event in note_events: if not isinstance(event, (list, tuple)) or len(event) < 4: continue try: start = float(event[0]) end = float(event[1]) pitch = int(round(float(event[2]))) velocity = float(event[3]) confidence = float(event[4]) if len(event) > 4 else 1.0 except (TypeError, ValueError): continue duracao = max(0.0, end - start) if strict: if duracao < 0.04: continue if confidence < 0.12: continue else: if duracao < 0.01: continue if confidence < 0.02: continue saida.append( { "start": start, "end": end, "duracao": duracao, "pitch": pitch, "velocity": velocity, "confidence": confidence, } ) return saida def extrair_nota_dominante(eventos: list[dict[str, float | int]]) -> tuple[Optional[int], float]: if not eventos: return None, 0.0 score_por_pitch: Counter[int] = Counter() total_score = 0.0 for evento in eventos: pitch = int(evento.get("pitch", evento.get("midi", 0))) duracao = evento.get("duracao") if duracao is None: duracao = max(0.0, float(evento.get("fim", 0.0)) - float(evento.get("inicio", 0.0))) score = float(duracao) * float(evento.get("confidence", evento.get("confianca", 1.0))) score_por_pitch[pitch] += score total_score += score if not score_por_pitch or total_score <= 0: return None, 0.0 pitch_dominante, score_dominante = score_por_pitch.most_common(1)[0] ratio = float(score_dominante) / float(total_score) return int(pitch_dominante), round(ratio, 4) def reconstruir_eventos_por_contorno(model_output: Any) -> list[dict[str, float | int]]: if not isinstance(model_output, dict): return [] contour = model_output.get("contour") if contour is None: return [] # Basic Pitch contour costuma ter 264 bins (3 bins por semitom, MIDI 21..108). # Aqui extraímos a nota dominante por frame e agrupamos em segmentos. frames: list[tuple[Optional[int], float]] = [] for row in contour: try: valores = list(row) except TypeError: continue if not valores: continue idx, conf = max(enumerate(valores), key=lambda x: float(x[1])) conf_f = float(conf) if conf_f < 0.12: frames.append((None, 0.0)) continue midi = int(round(21 + (idx / 3.0))) midi = max(21, min(108, midi)) frames.append((midi, conf_f)) if not frames: return [] hop_s = 0.023 # aproximacao estável para segmentação temporal min_segmento_frames = 2 eventos: list[dict[str, float | int]] = [] atual_pitch: Optional[int] = None atual_inicio = 0 confs: list[float] = [] def fechar_segmento(fim_idx: int) -> None: nonlocal atual_pitch, atual_inicio, confs if atual_pitch is None: return tamanho = fim_idx - atual_inicio if tamanho < min_segmento_frames: return inicio_s = atual_inicio * hop_s fim_s = fim_idx * hop_s duracao = max(0.0, fim_s - inicio_s) confidence = sum(confs) / max(1, len(confs)) eventos.append( { "start": inicio_s, "end": fim_s, "duracao": duracao, "pitch": int(atual_pitch), "velocity": 1.0, "confidence": float(confidence), } ) for idx, (pitch, conf) in enumerate(frames): if pitch is None: fechar_segmento(idx) atual_pitch = None confs = [] continue if atual_pitch is None: atual_pitch = pitch atual_inicio = idx confs = [conf] continue if abs(pitch - atual_pitch) <= 1: # suaviza tremulação de 1 semitom em áudio ambiente confs.append(conf) continue fechar_segmento(idx) atual_pitch = pitch atual_inicio = idx confs = [conf] fechar_segmento(len(frames)) return eventos class IdentificarRequest(BaseModel): path: Optional[str] = None texto: Optional[str] = None max_candidatos: int = 5 @app.post("/identificar") def identificar(req: IdentificarRequest): try: texto_referencia = normalizar_espacos(req.texto or "") transcricao = "" if texto_referencia: transcricao = texto_referencia else: if not req.path: raise HTTPException( status_code=422, detail="Envie path do wav ou texto para busca por letra.", ) wav_path = validar_arquivo_audio(req.path) transcricao = transcrever_wav_com_fallback(wav_path) if not transcricao: raise HTTPException( status_code=422, detail="Nao foi possivel transcrever o audio.", ) candidatos = buscar_candidatos_genius_por_letra( transcricao, max(1, min(req.max_candidatos, 10)), ) return { "transcricao": transcricao, "candidatos": candidatos, } except HTTPException: raise except Exception as e: traceback.print_exc() raise HTTPException(status_code=500, detail=str(e)) from e def validar_arquivo_audio(path_str: str) -> Path: path = Path(path_str or "").expanduser() if not path_str: raise HTTPException(status_code=422, detail="Path do audio nao informado.") if not path.exists() or not path.is_file(): raise HTTPException(status_code=404, detail=f"Arquivo nao encontrado: {path_str}") return path def transcrever_wav_com_fallback(wav_path: Path) -> str: prefer_local = parse_bool_env("LYRICS_PREFER_LOCAL_WHISPER", True) erros: list[str] = [] estrategias = ( [transcrever_wav_local_faster_whisper, transcrever_wav_openai] if prefer_local else [transcrever_wav_openai, transcrever_wav_local_faster_whisper] ) for estrategia in estrategias: try: texto = estrategia(wav_path) if texto: return texto except Exception as e: erros.append(f"{estrategia.__name__}: {e}") continue if erros: raise RuntimeError(" | ".join(erros)) return "" def transcrever_wav_openai(wav_path: Path) -> str: api_key = os.getenv("OPENAI_API_KEY", "").strip() if not api_key: raise RuntimeError("OPENAI_API_KEY nao configurada.") model = os.getenv("OPENAI_TRANSCRIBE_MODEL", "whisper-1").strip() or "whisper-1" max_seconds = parse_int_env("LYRICS_TRANSCRIBE_MAX_SECONDS", 45, min_value=10, max_value=300) clip_path = cortar_wav_prefixo(wav_path, max_seconds) headers = { "Authorization": f"Bearer {api_key}", } try: with open(clip_path, "rb") as audio_file: files = { "file": (clip_path.name, audio_file, "audio/wav"), } data = { "model": model, "response_format": "json", "temperature": "0", } response = requests.post( "https://api.openai.com/v1/audio/transcriptions", headers=headers, files=files, data=data, timeout=90, ) finally: if clip_path != wav_path and clip_path.exists(): try: clip_path.unlink() except OSError: pass if response.status_code < 200 or response.status_code >= 300: raise RuntimeError( f"Falha na transcricao OpenAI: HTTP {response.status_code} - {response.text[:400]}" ) payload: dict[str, Any] = response.json() texto = str(payload.get("text") or "").strip() return normalizar_espacos(texto) def transcrever_wav_local_faster_whisper(wav_path: Path) -> str: model = get_faster_whisper_model() max_seconds = parse_int_env("LYRICS_TRANSCRIBE_MAX_SECONDS", 45, min_value=10, max_value=300) clip_path = cortar_wav_prefixo(wav_path, max_seconds) beam_size = parse_int_env("LYRICS_WHISPER_BEAM_SIZE", 1, min_value=1, max_value=5) language = (os.getenv("LYRICS_WHISPER_LANGUAGE", "pt") or "").strip() or None try: segments, _info = model.transcribe( str(clip_path), language=language, beam_size=beam_size, vad_filter=True, condition_on_previous_text=False, ) partes = [normalizar_espacos(getattr(segment, "text", "")) for segment in segments] texto = normalizar_espacos(" ".join([p for p in partes if p])) if not texto: raise RuntimeError("Transcricao local vazia.") return texto finally: if clip_path != wav_path and clip_path.exists(): try: clip_path.unlink() except OSError: pass def get_faster_whisper_model(): global _FASTER_WHISPER_MODEL if _FASTER_WHISPER_MODEL is not None: return _FASTER_WHISPER_MODEL try: from faster_whisper import WhisperModel except Exception as e: raise RuntimeError( f"faster-whisper indisponivel (instale em requirements): {e}" ) from e model_size = (os.getenv("LYRICS_WHISPER_MODEL", "small") or "").strip() or "small" device = (os.getenv("LYRICS_WHISPER_DEVICE", "cpu") or "").strip() or "cpu" default_compute = "int8" if device == "cpu" else "float16" compute_type = (os.getenv("LYRICS_WHISPER_COMPUTE_TYPE", default_compute) or "").strip() _FASTER_WHISPER_MODEL = WhisperModel( model_size, device=device, compute_type=compute_type, ) return _FASTER_WHISPER_MODEL def buscar_candidatos_genius_por_letra(texto: str, max_candidatos: int) -> list[dict[str, str]]: token = os.getenv("GENIUS_ACCESS_TOKEN", "").strip() if not token: raise RuntimeError("GENIUS_ACCESS_TOKEN nao configurada.") queries = montar_queries_busca(texto) headers = {"Authorization": f"Bearer {token}"} candidatos_por_chave: dict[str, dict[str, Any]] = {} texto_norm = normalizar_texto_comparacao(texto) for query in queries: try: response = requests.get( "https://api.genius.com/search", params={"q": query}, headers=headers, timeout=20, ) if response.status_code < 200 or response.status_code >= 300: print(f"Genius search falhou (HTTP {response.status_code}) para query={query!r}") continue data = response.json() hits = data.get("response", {}).get("hits", []) for hit in hits: result = hit.get("result", {}) or {} titulo = str(result.get("title") or "").strip() primary_artist = result.get("primary_artist", {}) or {} artista = str(primary_artist.get("name") or "").strip() if not titulo or not artista: continue chave = f"{artista.lower()}::{titulo.lower()}" score = pontuar_candidato_titulo(texto_norm, titulo) existente = candidatos_por_chave.get(chave) if existente and existente.get("score", -1) >= score: continue candidatos_por_chave[chave] = { "titulo": titulo, "artista": artista, "score": score, } except Exception as e: print(f"Erro na busca Genius para query={query!r}: {e}") continue candidatos_ordenados = sorted( candidatos_por_chave.values(), key=lambda item: ( -int(item.get("score", 0)), len(str(item.get("titulo", ""))), str(item.get("artista", "")).lower(), str(item.get("titulo", "")).lower(), ), ) return [ {"titulo": str(item["titulo"]), "artista": str(item["artista"])} for item in candidatos_ordenados[:max_candidatos] ] def montar_queries_busca(texto: str) -> list[str]: tokens = [ token for token in re.split(r"\s+", normalizar_espacos(texto.lower())) if token ] if not tokens: return [] max_words = parse_int_env("LYRICS_SEARCH_WORDS", 10, min_value=6, max_value=20) if len(tokens) <= max_words: base = " ".join(tokens) queries = [base] if len(tokens) <= 5: queries.append(f'"{base}"') unicas: list[str] = [] vistos: set[str] = set() for q in queries: if q in vistos: continue vistos.add(q) unicas.append(q) return unicas inicio = tokens[:max_words] meio_start = max(0, (len(tokens) // 2) - (max_words // 2)) meio = tokens[meio_start : meio_start + max_words] fim = tokens[-max_words:] queries = [ " ".join(inicio), " ".join(meio), " ".join(fim), ] # Remove duplicadas preservando ordem. unicas: list[str] = [] vistos: set[str] = set() for q in queries: if q in vistos: continue vistos.add(q) unicas.append(q) return unicas def cortar_wav_prefixo(wav_path: Path, max_seconds: int) -> Path: with wave.open(str(wav_path), "rb") as src: channels = src.getnchannels() sample_width = src.getsampwidth() frame_rate = src.getframerate() total_frames = src.getnframes() if frame_rate <= 0: return wav_path max_frames = min(total_frames, int(frame_rate * max_seconds)) if max_frames >= total_frames: return wav_path frames = src.readframes(max_frames) with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp: clip_path = Path(tmp.name) with wave.open(str(clip_path), "wb") as dst: dst.setnchannels(channels) dst.setsampwidth(sample_width) dst.setframerate(frame_rate) dst.writeframes(frames) return clip_path def parse_int_env(name: str, default: int, min_value: int, max_value: int) -> int: raw = os.getenv(name, "").strip() if not raw: return default try: value = int(raw) except ValueError: return default return max(min_value, min(max_value, value)) def parse_bool_env(name: str, default: bool) -> bool: raw = os.getenv(name, "").strip().lower() if not raw: return default if raw in {"1", "true", "yes", "y", "on"}: return True if raw in {"0", "false", "no", "n", "off"}: return False return default def normalizar_espacos(texto: str) -> str: return re.sub(r"\s+", " ", texto or "").strip() def normalizar_texto_comparacao(texto: str) -> str: texto = normalizar_espacos((texto or "").lower()) texto = re.sub(r"[^a-z0-9\s]+", " ", texto) return normalizar_espacos(texto) def pontuar_candidato_titulo(texto_referencia_norm: str, titulo: str) -> int: titulo_norm = normalizar_texto_comparacao(titulo) if not texto_referencia_norm or not titulo_norm: return 0 score = 0 if titulo_norm == texto_referencia_norm: score += 1000 elif titulo_norm.startswith(texto_referencia_norm): score += 800 elif texto_referencia_norm in titulo_norm: score += 650 elif titulo_norm in texto_referencia_norm: score += 500 ref_tokens = [t for t in texto_referencia_norm.split(" ") if t] titulo_tokens = [t for t in titulo_norm.split(" ") if t] if ref_tokens and titulo_tokens: ref_set = set(ref_tokens) titulo_set = set(titulo_tokens) inter = len(ref_set & titulo_set) if inter: score += inter * 40 score += int((inter / max(1, len(ref_set))) * 100) return score