python-basic-pitch/report_audio_accuracy.py

from __future__ import annotations

import argparse
import json
import os
import re
import sys
import time
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any, Iterable

sys.path.insert(0, str(Path(__file__).resolve().parent))

from music_token_metrics import (
    MatrizSemAcorde,
    VERSAO_ESQUEMA_METRICAS,
    acuracia_posicional,
    avaliar_sem_acorde,
    distancia_edicao_tokens,
    melhor_rotacao,
    normalizar_token,
    qualidade_completa_token,
    qualidade_token,
    raiz_token,
    simplificar_token_musical,
    tokenizar,
)


RAIZ_PROJETO = Path(__file__).resolve().parents[1]
CAMINHO_DESCRICAO_PADRAO = RAIZ_PROJETO / "audios-testes" / "descricao.md"
FAIXAS_POR_INSTRUMENTO = {
    "violao": (80.0, 1200.0),
    "teclado": (27.5, 4200.0),
    "sax_alto": (130.0, 900.0),
    "violino": (196.0, 3500.0),
}


@dataclass(frozen=True)
class CasoAudio:
    nome_arquivo: str
    esperado: str
    instrumento: str
    caminho: Path


@dataclass(frozen=True)
class ResultadoAudio:
    filename: str
    instrument: str
    expected: str
    predicted: str
    exact_match: bool
    slash_tolerated_match: bool
    rotated_exact_match: bool
    rotated_simplified_match: bool
    rotated_root_match: bool
    musical_match: bool
    token_accuracy: float
    simplified_token_accuracy: float
    root_accuracy: float
    quality_accuracy: float
    full_quality_accuracy: float
    edit_distance: int
    substitutions: int
    insertions: int
    deletions: int
    token_error_rate: float
    no_chord_true_positives: int
    no_chord_false_positives: int
    no_chord_false_negatives: int
    no_chord_true_negatives: int
    best_rotation_shift: int
    expected_tokens: int
    predicted_tokens: int
    matched_tokens: int
    elapsed_seconds: float
    diagnostics: dict[str, Any]


def ler_argumentos() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Relatorio de assertividade com audios reais.")
    parser.add_argument("--spec", type=Path, default=CAMINHO_DESCRICAO_PADRAO, help="Arquivo descricao.md com saidas esperadas.")
    parser.add_argument("--json", action="store_true", help="Imprime o relatorio em JSON.")
    parser.add_argument(
        "--instrument",
        action="append",
        default=[],
        help="Filtra por instrumento (ex.: --instrument violao --instrument teclado).",
    )
    parser.add_argument(
        "--no-fixtures",
        action="store_true",
        help="Desliga as fixtures de regressao para medir somente a saida real do algoritmo.",
    )
    parser.add_argument(
        "--verbose",
        action="store_true",
        help="Imprime progresso e duracao por caso conforme a analise acontece.",
    )
    parser.add_argument(
        "--case",
        action="append",
        default=[],
        help="Filtra por trecho do nome do arquivo (ex.: --case bia --case teclado).",
    )
    parser.add_argument(
        "--limit",
        type=int,
        default=0,
        help="Limita a quantidade de casos processados apos os filtros.",
    )
    parser.add_argument(
        "--validate-corpus",
        action="store_true",
        help="Valida descricao e arquivos sem executar o pipeline DSP.",
    )
    return parser.parse_args()


def carregar_casos(caminho_descricao: Path) -> list[CasoAudio]:
    conteudo = ler_texto_com_fallback(caminho_descricao)
    padrao = re.compile(
        "^Audio:\\s*(.+?)\\r?\\n"
        "^Sa(?:ida|\u00edda|\u00c3\u00adda) esperada:\\s*(.+?)(?:\\r?\\n\\s*=+|\\Z)",
        re.IGNORECASE | re.DOTALL | re.MULTILINE,
    )
    casos: list[CasoAudio] = []
    for nome_arquivo, esperado in padrao.findall(conteudo):
        nome_normalizado = nome_arquivo.strip()
        caminho_audio = resolver_caminho_audio(caminho_descricao.parent, nome_normalizado)
        casos.append(
            CasoAudio(
                nome_arquivo=nome_normalizado,
                esperado=normalizar_saida_esperada(esperado),
                instrumento=inferir_instrumento(nome_normalizado),
                caminho=caminho_audio,
            )
        )
    return casos


def ler_texto_com_fallback(caminho: Path) -> str:
    for encoding in ("utf-8", "utf-8-sig", "cp1252", "latin-1"):
        try:
            return caminho.read_text(encoding=encoding)
        except UnicodeDecodeError:
            continue
    return caminho.read_text(encoding="utf-8", errors="replace")


def resolver_caminho_audio(pasta: Path, nome_arquivo: str) -> Path:
    caminho_direto = pasta / nome_arquivo
    if caminho_direto.exists():
        return caminho_direto

    nome_base_esperado = Path(nome_arquivo).stem.casefold()
    encontrados = [
        caminho
        for caminho in pasta.iterdir()
        if caminho.is_file() and caminho.stem.casefold() == nome_base_esperado
    ]
    if len(encontrados) == 1:
        return encontrados[0]

    raise FileNotFoundError(f"Arquivo nao encontrado para '{nome_arquivo}' em {pasta}")


def inferir_instrumento(nome_arquivo: str) -> str:
    texto = nome_arquivo.casefold()
    if "sax" in texto:
        return "sax_alto"
    if "violino" in texto:
        return "violino"
    if "teclado" in texto:
        return "teclado"
    return "violao"


def normalizar_saida_esperada(valor: str) -> str:
    return " ".join(tokenizar(valor))


def resumo_diagnostico(resultado: dict[str, Any]) -> dict[str, Any]:
    diagnostico_harmonico = resultado.get("diagnostico_harmonico") or {}
    return {
        "tom": resultado.get("tom"),
        "modo": resultado.get("modo"),
        "confianca_tom": resultado.get("confianca_tom"),
        "perfil_fonte_audio": resultado.get("perfil_fonte_audio"),
        "base_harmonica_auxiliar": resultado.get("base_harmonica_auxiliar"),
        "tonalidades_candidatas": resultado.get("tonalidades_candidatas", [])[:3],
        "candidatas_progressao": resultado.get("candidatas_progressao", [])[:3],
        "chord_candidates": resultado.get("chord_candidates", [])[:4],
        "no_chord_detected": diagnostico_harmonico.get("no_chord_detected", False),
        "no_chord_probability": diagnostico_harmonico.get("no_chord_probability"),
        "no_chord_threshold": diagnostico_harmonico.get("no_chord_threshold"),
        "no_chord_reasons": diagnostico_harmonico.get("no_chord_reasons", []),
        "score_kind": diagnostico_harmonico.get("score_kind"),
        "timings": resultado.get("timings", {}),
    }


def analisar_caso(caso: CasoAudio) -> ResultadoAudio:
    inicio = time.perf_counter()
    diagnostics: dict[str, Any] = {}
    if caso.instrumento in {"sax_alto", "violino"}:
        import main as pitch_main

        resultado = pitch_main.analisar_melodico(caso.caminho, caso.instrumento)
        tokens_obtidos = [normalizar_token(token) for token in resultado.get("notas_resumo", [])]
    else:
        from harmonic_pipeline import analyze_harmonic_audio

        resultado = analyze_harmonic_audio(
            str(caso.caminho),
            caso.instrumento,
            FAIXAS_POR_INSTRUMENTO[caso.instrumento],
        )
        tokens_obtidos = tokenizar(resultado.get("cifra_palco", ""))
        diagnostico_harmonico = resultado.get("diagnostico_harmonico") or {}
        if diagnostico_harmonico.get("no_chord_detected") is True:
            # N e um estado acustico, nao uma cifra. Convertemos para token
            # apenas na fronteira do benchmark para medir TP/FP/FN corretamente.
            tokens_obtidos = ["N"]
        diagnostics = resumo_diagnostico(resultado)

    tokens_esperados = tokenizar(caso.esperado)
    tokens_corretos, token_accuracy = acuracia_posicional(tokens_esperados, tokens_obtidos)

    simplificados_esperados = [simplificar_token_musical(token) for token in tokens_esperados]
    simplificados_obtidos = [simplificar_token_musical(token) for token in tokens_obtidos]
    _, simplified_token_accuracy = acuracia_posicional(simplificados_esperados, simplificados_obtidos)

    raizes_esperadas = [raiz_token(token) for token in tokens_esperados]
    raizes_obtidas = [raiz_token(token) for token in tokens_obtidos]
    _, root_accuracy = acuracia_posicional(raizes_esperadas, raizes_obtidas)

    qualidades_esperadas = [qualidade_token(token) for token in tokens_esperados]
    qualidades_obtidas = [qualidade_token(token) for token in tokens_obtidos]
    _, quality_accuracy = acuracia_posicional(qualidades_esperadas, qualidades_obtidas)

    qualidades_completas_esperadas = [qualidade_completa_token(token) for token in tokens_esperados]
    qualidades_completas_obtidas = [qualidade_completa_token(token) for token in tokens_obtidos]
    _, full_quality_accuracy = acuracia_posicional(
        qualidades_completas_esperadas,
        qualidades_completas_obtidas,
    )
    edicao = distancia_edicao_tokens(tokens_esperados, tokens_obtidos)
    matriz_sem_acorde = avaliar_sem_acorde(tokens_esperados, tokens_obtidos)

    exact_match = tokens_esperados == tokens_obtidos
    slash_tolerated_match = simplificados_esperados == simplificados_obtidos
    rotated_exact_match, _, exact_shift = melhor_rotacao(tokens_esperados, tokens_obtidos, lambda valor: valor)
    rotated_simplified_match, _, simplified_shift = melhor_rotacao(
        tokens_esperados,
        tokens_obtidos,
        simplificar_token_musical,
    )
    rotated_root_match, _, root_shift = melhor_rotacao(tokens_esperados, tokens_obtidos, raiz_token)
    musical_match = exact_match or slash_tolerated_match or rotated_simplified_match

    return ResultadoAudio(
        filename=caso.caminho.name,
        instrument=caso.instrumento,
        expected=" ".join(tokens_esperados),
        predicted=" ".join(tokens_obtidos),
        exact_match=exact_match,
        slash_tolerated_match=slash_tolerated_match,
        rotated_exact_match=rotated_exact_match,
        rotated_simplified_match=rotated_simplified_match,
        rotated_root_match=rotated_root_match,
        musical_match=musical_match,
        token_accuracy=token_accuracy,
        simplified_token_accuracy=simplified_token_accuracy,
        root_accuracy=root_accuracy,
        quality_accuracy=quality_accuracy,
        full_quality_accuracy=full_quality_accuracy,
        edit_distance=edicao.distancia,
        substitutions=edicao.substituicoes,
        insertions=edicao.insercoes,
        deletions=edicao.remocoes,
        token_error_rate=edicao.distancia / max(len(tokens_esperados), 1),
        no_chord_true_positives=matriz_sem_acorde.verdadeiros_positivos,
        no_chord_false_positives=matriz_sem_acorde.falsos_positivos,
        no_chord_false_negatives=matriz_sem_acorde.falsos_negativos,
        no_chord_true_negatives=matriz_sem_acorde.verdadeiros_negativos,
        best_rotation_shift=exact_shift or simplified_shift or root_shift,
        expected_tokens=len(tokens_esperados),
        predicted_tokens=len(tokens_obtidos),
        matched_tokens=tokens_corretos,
        elapsed_seconds=time.perf_counter() - inicio,
        diagnostics=diagnostics,
    )


def resumir_resultados(resultados: Iterable[ResultadoAudio]) -> dict[str, Any]:
    lista = list(resultados)
    total = len(lista)
    acertos_exatos = sum(1 for item in lista if item.exact_match)
    acertos_musicais = sum(1 for item in lista if item.musical_match)
    por_instrumento: dict[str, dict[str, Any]] = {}

    for item in lista:
        resumo = por_instrumento.setdefault(
            item.instrument,
            {
                "cases": 0,
                "exact_matches": 0,
                "musical_matches": 0,
                "mean_token_accuracy": 0.0,
                "mean_simplified_accuracy": 0.0,
                "mean_root_accuracy": 0.0,
                "mean_quality_accuracy": 0.0,
                "mean_full_quality_accuracy": 0.0,
                "mean_elapsed_seconds": 0.0,
                "edit_distance": 0,
                "substitutions": 0,
                "insertions": 0,
                "deletions": 0,
                "expected_tokens": 0,
                "no_chord_true_positives": 0,
                "no_chord_false_positives": 0,
                "no_chord_false_negatives": 0,
                "no_chord_true_negatives": 0,
            },
        )
        resumo["cases"] += 1
        resumo["exact_matches"] += 1 if item.exact_match else 0
        resumo["musical_matches"] += 1 if item.musical_match else 0
        resumo["mean_token_accuracy"] += item.token_accuracy
        resumo["mean_simplified_accuracy"] += item.simplified_token_accuracy
        resumo["mean_root_accuracy"] += item.root_accuracy
        resumo["mean_quality_accuracy"] += item.quality_accuracy
        resumo["mean_full_quality_accuracy"] += item.full_quality_accuracy
        resumo["mean_elapsed_seconds"] += item.elapsed_seconds
        resumo["edit_distance"] += item.edit_distance
        resumo["substitutions"] += item.substitutions
        resumo["insertions"] += item.insertions
        resumo["deletions"] += item.deletions
        resumo["expected_tokens"] += item.expected_tokens
        resumo["no_chord_true_positives"] += item.no_chord_true_positives
        resumo["no_chord_false_positives"] += item.no_chord_false_positives
        resumo["no_chord_false_negatives"] += item.no_chord_false_negatives
        resumo["no_chord_true_negatives"] += item.no_chord_true_negatives

    for resumo in por_instrumento.values():
        quantidade = max(int(resumo["cases"]), 1)
        resumo["mean_token_accuracy"] = resumo["mean_token_accuracy"] / quantidade
        resumo["mean_simplified_accuracy"] = resumo["mean_simplified_accuracy"] / quantidade
        resumo["mean_root_accuracy"] = resumo["mean_root_accuracy"] / quantidade
        resumo["mean_quality_accuracy"] = resumo["mean_quality_accuracy"] / quantidade
        resumo["mean_full_quality_accuracy"] = resumo["mean_full_quality_accuracy"] / quantidade
        resumo["mean_elapsed_seconds"] = resumo["mean_elapsed_seconds"] / quantidade
        resumo["token_error_rate"] = resumo["edit_distance"] / max(resumo["expected_tokens"], 1)
        resumo["no_chord"] = resumir_matriz_sem_acorde(
            MatrizSemAcorde(
                verdadeiros_positivos=resumo.pop("no_chord_true_positives"),
                falsos_positivos=resumo.pop("no_chord_false_positives"),
                falsos_negativos=resumo.pop("no_chord_false_negatives"),
                verdadeiros_negativos=resumo.pop("no_chord_true_negatives"),
            )
        )

    matriz_sem_acorde_global = MatrizSemAcorde()
    for item in lista:
        matriz_sem_acorde_global = matriz_sem_acorde_global.somar(
            MatrizSemAcorde(
                verdadeiros_positivos=item.no_chord_true_positives,
                falsos_positivos=item.no_chord_false_positives,
                falsos_negativos=item.no_chord_false_negatives,
                verdadeiros_negativos=item.no_chord_true_negatives,
            )
        )
    total_tokens_esperados = sum(item.expected_tokens for item in lista)
    distancia_edicao_total = sum(item.edit_distance for item in lista)

    return {
        "metrics_schema_version": VERSAO_ESQUEMA_METRICAS,
        "fixtures_enabled": fixtures_habilitadas(),
        "total_cases": total,
        "exact_matches": acertos_exatos,
        "exact_match_rate": (acertos_exatos / total) if total else 0.0,
        "musical_matches": acertos_musicais,
        "musical_match_rate": (acertos_musicais / total) if total else 0.0,
        "mean_token_accuracy": (sum(item.token_accuracy for item in lista) / total) if total else 0.0,
        "mean_simplified_accuracy": (sum(item.simplified_token_accuracy for item in lista) / total) if total else 0.0,
        "mean_root_accuracy": (sum(item.root_accuracy for item in lista) / total) if total else 0.0,
        "mean_quality_accuracy": (sum(item.quality_accuracy for item in lista) / total) if total else 0.0,
        "mean_full_quality_accuracy": (
            sum(item.full_quality_accuracy for item in lista) / total
        ) if total else 0.0,
        "token_error_rate": distancia_edicao_total / max(total_tokens_esperados, 1),
        "edit_operations": {
            "distance": distancia_edicao_total,
            "substitutions": sum(item.substitutions for item in lista),
            "insertions": sum(item.insertions for item in lista),
            "deletions": sum(item.deletions for item in lista),
            "expected_tokens": total_tokens_esperados,
        },
        "no_chord": resumir_matriz_sem_acorde(matriz_sem_acorde_global),
        "mean_elapsed_seconds": (sum(item.elapsed_seconds for item in lista) / total) if total else 0.0,
        "by_instrument": por_instrumento,
        "results": [asdict(item) for item in lista],
    }


def resumir_matriz_sem_acorde(matriz: MatrizSemAcorde) -> dict[str, Any]:
    return {
        "support": matriz.suporte,
        "true_positives": matriz.verdadeiros_positivos,
        "false_positives": matriz.falsos_positivos,
        "false_negatives": matriz.falsos_negativos,
        "true_negatives": matriz.verdadeiros_negativos,
        "accuracy": matriz.acuracia,
        "precision": matriz.precisao,
        "recall": matriz.revocacao,
        "f1": matriz.f1,
        "zero_division": 0,
    }


def imprimir_relatorio_humano(resumo: dict[str, Any]) -> None:
    modo = "com fixtures de regressao" if resumo["fixtures_enabled"] else "sem fixtures"
    print("Relatorio de Assertividade")
    print(f"Esquema de metricas: {resumo['metrics_schema_version']}")
    print(f"Modo: {modo}")
    print(f"Casos: {resumo['total_cases']}")
    print(f"Acerto exato: {resumo['exact_matches']}/{resumo['total_cases']} ({resumo['exact_match_rate']:.1%})")
    print(f"Acerto musical: {resumo['musical_matches']}/{resumo['total_cases']} ({resumo['musical_match_rate']:.1%})")
    print(f"Media por token: {resumo['mean_token_accuracy']:.1%}")
    print(f"Media sem baixo/ornamento: {resumo['mean_simplified_accuracy']:.1%}")
    print(f"Media de raiz: {resumo['mean_root_accuracy']:.1%}")
    print(f"Media de qualidade: {resumo['mean_quality_accuracy']:.1%}")
    print(f"Media de qualidade completa: {resumo['mean_full_quality_accuracy']:.1%}")
    operacoes = resumo["edit_operations"]
    print(
        f"Taxa de erro sequencial: {resumo['token_error_rate']:.1%} "
        f"(S={operacoes['substitutions']}, I={operacoes['insertions']}, D={operacoes['deletions']})"
    )
    no_chord = resumo["no_chord"]
    print(
        f"N/no-chord: suporte={no_chord['support']}, FP={no_chord['false_positives']}, "
        f"FN={no_chord['false_negatives']}, precisao={no_chord['precision']:.1%}, "
        f"revocacao={no_chord['recall']:.1%}, F1={no_chord['f1']:.1%}"
    )
    print(f"Tempo medio por caso: {resumo['mean_elapsed_seconds']:.2f}s")
    print("")
    print("Por instrumento:")
    for instrumento, dados in sorted(resumo["by_instrument"].items()):
        print(
            f"- {instrumento}: {int(dados['exact_matches'])}/{int(dados['cases'])} exatos "
            f"({(dados['exact_matches'] / max(dados['cases'], 1)):.1%}), "
            f"musical {int(dados['musical_matches'])}/{int(dados['cases'])} "
            f"({(dados['musical_matches'] / max(dados['cases'], 1)):.1%}), "
            f"raiz {(dados['mean_root_accuracy']):.1%}, "
            f"qualidade completa {(dados['mean_full_quality_accuracy']):.1%}, "
            f"TER {(dados['token_error_rate']):.1%}"
        )
    print("")
    print("Casos:")
    for item in resumo["results"]:
        status = "OK" if item["exact_match"] else ("MUSICAL" if item["musical_match"] else "FALHOU")
        total_tokens = max(item["expected_tokens"], item["predicted_tokens"], 1)
        print(f"- {status} [{item['instrument']}] {item['filename']}")
        print(f"  esperado: {item['expected']}")
        print(f"  obtido:   {item['predicted']}")
        print(f"  tokens:   {item['matched_tokens']}/{total_tokens} ({item['token_accuracy']:.1%})")
        print(
            f"  musical:  sem-baixo {str(item['slash_tolerated_match']).lower()}, "
            f"rotacao {str(item['rotated_simplified_match']).lower()}, "
            f"raiz {item['root_accuracy']:.1%}, qualidade {item['quality_accuracy']:.1%}, "
            f"qualidade completa {item['full_quality_accuracy']:.1%}"
        )
        print(
            f"  edicao:   TER {item['token_error_rate']:.1%}, "
            f"S={item['substitutions']}, I={item['insertions']}, D={item['deletions']}"
        )
        print(f"  tempo:    {item['elapsed_seconds']:.2f}s")
        if not item["exact_match"] and item.get("diagnostics"):
            diagnostico = item["diagnostics"]
            print(
                f"  diag:     tom {diagnostico.get('tom') or '-'} {diagnostico.get('modo') or ''} "
                f"conf {diagnostico.get('confianca_tom') or '-'}"
            )
            candidatos = diagnostico.get("candidatas_progressao") or []
            if candidatos:
                resumo_candidatos = [
                    " ".join(candidato.get("acordes", [])[:6])
                    for candidato in candidatos[:2]
                ]
                print(f"  cand:     {' | '.join(resumo_candidatos)}")


def analisar_casos(casos: list[CasoAudio], verbose: bool = False) -> list[ResultadoAudio]:
    resultados: list[ResultadoAudio] = []
    total = len(casos)
    for indice, caso in enumerate(casos, start=1):
        if verbose:
            print(f"[{indice}/{total}] analisando [{caso.instrumento}] {caso.caminho.name}...", flush=True)
        resultado = analisar_caso(caso)
        resultados.append(resultado)
        if verbose:
            status = "OK" if resultado.exact_match else ("MUSICAL" if resultado.musical_match else "FALHOU")
            print(
                f"[{indice}/{total}] {status} em {resultado.elapsed_seconds:.2f}s | "
                f"{resultado.filename} | token {resultado.token_accuracy:.1%}",
                flush=True,
            )
    return resultados


def filtrar_casos(casos: list[CasoAudio], args: argparse.Namespace) -> list[CasoAudio]:
    filtrados = list(casos)
    if args.instrument:
        instrumentos = {valor.strip().lower() for valor in args.instrument if valor.strip()}
        filtrados = [caso for caso in filtrados if caso.instrumento.lower() in instrumentos]
    if args.case:
        termos = [valor.strip().casefold() for valor in args.case if valor.strip()]
        filtrados = [
            caso
            for caso in filtrados
            if any(termo in caso.nome_arquivo.casefold() for termo in termos)
        ]
    if args.limit > 0:
        filtrados = filtrados[: args.limit]
    return filtrados


def fixtures_habilitadas() -> bool:
    valor = str(os.getenv("AUDIO_REGRESSION_FIXTURES", "0")).strip().lower()
    return valor in {"1", "true", "on", "yes"}


def main() -> int:
    args = ler_argumentos()
    if args.no_fixtures:
        os.environ["AUDIO_REGRESSION_FIXTURES"] = "0"

    casos = filtrar_casos(carregar_casos(args.spec), args)
    if not casos:
        raise RuntimeError(f"Nenhum caso de audio valido encontrado em {args.spec}")
    if args.validate_corpus:
        print(f"Corpus valido: {len(casos)} casos em {args.spec}")
        return 0
    resultados = analisar_casos(casos, verbose=args.verbose)
    resumo = resumir_resultados(resultados)

    if args.json:
        print(json.dumps(resumo, ensure_ascii=False, indent=2))
    else:
        imprimir_relatorio_humano(resumo)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())