python-basic-pitch/main.py

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