feat: carregamento automático da análise do Scanner ao abrir a aba
- carregamento automático da análise do Scanner ao abrir a aba Refinar - simplificada a aba Editar vídeo para exibir análises disponíveis sem carregamento manual - movida a visualização do tipo de vídeo para a guia Projeto com leitura do banco Resumo: - 10 arquivos alterados - 2 novos - 8 modificados - 0 removidos 8 files changed, 114 insertions(+), 11 deletions(-) Arquivos: - .jhonny/analises.db - code/cep-plugin/abas/projeto/projeto.js - code/cep-plugin/index.html - code/cep-plugin/main.js - code/engine/integracoes/audio/__init__.py - code/engine/integracoes/audio/modelos_de_analise_musical.py - code/engine/integracoes/audio/provider_de_analise_musical_essentia.py - code/engine/testes/test_analisador_de_musica_essentia.py - code/engine/consultar_edicao_concluida.py - code/engine/integracoes/audio/classificador_de_musica_essentia.py
This commit is contained in:
João Henrique
@@ -0,0 +1,56 @@
|
||||
"""Consulta a configuração editorial concluída mais recente de um vídeo."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
import sqlite3
|
||||
import sys
|
||||
|
||||
CAMINHO_DO_CODIGO = Path(__file__).resolve().parent.parent
|
||||
if str(CAMINHO_DO_CODIGO) not in sys.path:
|
||||
sys.path.insert(0, str(CAMINHO_DO_CODIGO))
|
||||
|
||||
from engine.persistencia.conexao import abrir_banco
|
||||
|
||||
|
||||
def consultar(caminho_do_banco: Path, video_id: str) -> dict[str, object] | None:
|
||||
"""Retorna a edição mais recente sem alterar o banco."""
|
||||
conexao = abrir_banco(caminho_do_banco)
|
||||
try:
|
||||
linha = conexao.execute(
|
||||
"""SELECT id, video_id, sequencia, tipo_de_video, configuracao, concluida_em
|
||||
FROM edicoes_de_video WHERE video_id = ?
|
||||
ORDER BY concluida_em DESC, id DESC LIMIT 1""", (video_id,),
|
||||
).fetchone()
|
||||
if linha is None:
|
||||
return None
|
||||
return {
|
||||
"id": linha["id"], "video_id": linha["video_id"],
|
||||
"sequencia": linha["sequencia"], "tipo_de_video": linha["tipo_de_video"],
|
||||
"configuracao": json.loads(linha["configuracao"]),
|
||||
"concluida_em": linha["concluida_em"],
|
||||
}
|
||||
finally:
|
||||
conexao.close()
|
||||
|
||||
|
||||
def principal() -> None:
|
||||
"""Executa a consulta pela linha de comando."""
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("video_id")
|
||||
parser.add_argument("--banco", type=Path, default=Path(".jhonny/analises.db"))
|
||||
opcoes = parser.parse_args()
|
||||
try:
|
||||
edicao = consultar(opcoes.banco, opcoes.video_id)
|
||||
if edicao is None:
|
||||
raise ValueError("Nenhuma edição concluída encontrada para este vídeo.")
|
||||
print(json.dumps({"ok": True, "edicao": edicao}, ensure_ascii=False))
|
||||
except (OSError, ValueError, sqlite3.Error, json.JSONDecodeError) as erro:
|
||||
print(json.dumps({"ok": False, "erro": str(erro)}, ensure_ascii=False))
|
||||
raise SystemExit(1) from erro
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
principal()
|
||||
@@ -1,11 +1,14 @@
|
||||
"""Integrações locais para análise de áudio e música."""
|
||||
|
||||
from .analisador_de_metricas_de_voz import AnalisadorDeMetricasDeVoz
|
||||
from .modelos_de_analise_musical import ResultadoDaAnaliseMusical
|
||||
from .classificador_de_musica_essentia import ClassificadorDeMusicaEssentia
|
||||
from .modelos_de_analise_musical import PontuacaoDeEtiqueta, ResultadoDaAnaliseMusical
|
||||
from .provider_de_analise_musical_essentia import AnalisadorDeMusicaInstrumentalEssentia
|
||||
|
||||
__all__ = [
|
||||
"AnalisadorDeMetricasDeVoz",
|
||||
"ClassificadorDeMusicaEssentia",
|
||||
"PontuacaoDeEtiqueta",
|
||||
"AnalisadorDeMusicaInstrumentalEssentia",
|
||||
"ResultadoDaAnaliseMusical",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
"""
|
||||
Classificação semântica de música por modelos oficiais do Essentia.
|
||||
|
||||
Este módulo carrega os pesos de gênero e humor, extrai um embedding de áudio
|
||||
uma única vez e aplica os classificadores sobre esse embedding. Os arquivos
|
||||
dos modelos ficam fora do repositório e são informados por configuração.
|
||||
"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from .modelos_de_analise_musical import PontuacaoDeEtiqueta
|
||||
|
||||
|
||||
class ClassificadorDeMusicaEssentia:
|
||||
"""
|
||||
Classifica gênero e humor diretamente do sinal de áudio.
|
||||
|
||||
A classe não lê tags do arquivo, não produz texto narrativo e não
|
||||
persiste resultados. Ela depende dos pesos ``.pb`` e metadados ``.json``
|
||||
oficiais do Essentia, recebidos por caminho.
|
||||
"""
|
||||
|
||||
NOME_DO_MODELO_DE_EMBEDDING = "discogs-effnet-bs64-1.pb"
|
||||
NOME_DO_MODELO_DE_GENERO = "genre_discogs400-discogs-effnet-1.pb"
|
||||
NOME_DOS_METADADOS_DE_GENERO = "genre_discogs400-discogs-effnet-1.json"
|
||||
NOME_DO_MODELO_DE_HUMOR = "mtg_jamendo_moodtheme-discogs-effnet-1.pb"
|
||||
NOME_DOS_METADADOS_DE_HUMOR = "mtg_jamendo_moodtheme-discogs-effnet-1.json"
|
||||
|
||||
def __init__(self, diretorio_dos_modelos: str | Path) -> None:
|
||||
"""Inicializa o classificador com o diretório dos pesos locais."""
|
||||
self._diretorio_dos_modelos = Path(diretorio_dos_modelos).expanduser().resolve()
|
||||
self._modelos_carregados: tuple[Any, Any, Any] | None = None
|
||||
self._classes_por_tarefa: dict[str, tuple[str, ...]] | None = None
|
||||
|
||||
def classificar(
|
||||
self, arquivo: str | Path,
|
||||
) -> tuple[tuple[PontuacaoDeEtiqueta, ...], tuple[PontuacaoDeEtiqueta, ...]]:
|
||||
"""
|
||||
Classifica o gênero e o humor diretamente do conteúdo do arquivo.
|
||||
|
||||
Parâmetros:
|
||||
arquivo: Caminho de áudio que será lido pelo Essentia.
|
||||
|
||||
Retorna:
|
||||
Tupla com pontuações de gênero e de humor, em ordem decrescente.
|
||||
|
||||
Pode gerar:
|
||||
FileNotFoundError: quando o arquivo não existir.
|
||||
RuntimeError: quando Essentia ou algum modelo não estiver disponível.
|
||||
"""
|
||||
caminho = Path(arquivo).expanduser().resolve(strict=False)
|
||||
if not caminho.is_file():
|
||||
raise FileNotFoundError(f"Arquivo de áudio não encontrado: {caminho}")
|
||||
try:
|
||||
from essentia.standard import (
|
||||
MonoLoader,
|
||||
TensorflowPredict2D,
|
||||
TensorflowPredictEffnetDiscogs,
|
||||
)
|
||||
except ImportError as erro:
|
||||
raise RuntimeError(
|
||||
"A classificação musical precisa do suporte TensorFlow do Essentia."
|
||||
) from erro
|
||||
|
||||
modelo_de_embedding, modelo_de_genero, modelo_de_humor = self._obter_modelos(
|
||||
TensorflowPredictEffnetDiscogs, TensorflowPredict2D,
|
||||
)
|
||||
audio = MonoLoader(filename=str(caminho), sampleRate=16000, resampleQuality=4)()
|
||||
embedding = modelo_de_embedding(audio)
|
||||
previsao_de_genero = modelo_de_genero(embedding)
|
||||
previsao_de_humor = modelo_de_humor(embedding)
|
||||
return (
|
||||
self._converter_pontuacoes(previsao_de_genero, "genero"),
|
||||
self._converter_pontuacoes(previsao_de_humor, "humor"),
|
||||
)
|
||||
|
||||
def _obter_modelos(self, classe_de_embedding: Any, classe_de_classificacao: Any):
|
||||
if self._modelos_carregados is not None:
|
||||
return self._modelos_carregados
|
||||
caminhos = {
|
||||
"embedding": self._diretorio_dos_modelos / self.NOME_DO_MODELO_DE_EMBEDDING,
|
||||
"genero": self._diretorio_dos_modelos / self.NOME_DO_MODELO_DE_GENERO,
|
||||
"humor": self._diretorio_dos_modelos / self.NOME_DO_MODELO_DE_HUMOR,
|
||||
}
|
||||
ausentes = [str(caminho) for caminho in caminhos.values() if not caminho.is_file()]
|
||||
if ausentes:
|
||||
raise RuntimeError("Modelos Essentia ausentes: " + ", ".join(ausentes))
|
||||
self._modelos_carregados = (
|
||||
classe_de_embedding(
|
||||
graphFilename=str(caminhos["embedding"]), output="PartitionedCall:1",
|
||||
),
|
||||
classe_de_classificacao(
|
||||
graphFilename=str(caminhos["genero"]),
|
||||
input="serving_default_model_Placeholder", output="PartitionedCall:0",
|
||||
),
|
||||
classe_de_classificacao(graphFilename=str(caminhos["humor"])),
|
||||
)
|
||||
self._classes_por_tarefa = {
|
||||
"genero": self._carregar_classes(self.NOME_DOS_METADADOS_DE_GENERO),
|
||||
"humor": self._carregar_classes(self.NOME_DOS_METADADOS_DE_HUMOR),
|
||||
}
|
||||
return self._modelos_carregados
|
||||
|
||||
def _carregar_classes(self, nome_do_arquivo: str) -> tuple[str, ...]:
|
||||
caminho = self._diretorio_dos_modelos / nome_do_arquivo
|
||||
if not caminho.is_file():
|
||||
raise RuntimeError(f"Metadados Essentia ausentes: {caminho}")
|
||||
try:
|
||||
dados = json.loads(caminho.read_text(encoding="utf-8"))
|
||||
classes = dados["classes"]
|
||||
except (OSError, json.JSONDecodeError, KeyError, TypeError) as erro:
|
||||
raise RuntimeError(f"Metadados Essentia inválidos: {caminho}") from erro
|
||||
if not isinstance(classes, list) or not all(isinstance(item, str) for item in classes):
|
||||
raise RuntimeError(f"Lista de classes Essentia inválida: {caminho}")
|
||||
return tuple(classes)
|
||||
|
||||
def _converter_pontuacoes(self, previsoes: Any, tarefa: str) -> tuple[PontuacaoDeEtiqueta, ...]:
|
||||
import numpy as np
|
||||
|
||||
if self._classes_por_tarefa is None:
|
||||
raise RuntimeError("As classes dos modelos ainda não foram carregadas.")
|
||||
valores = np.asarray(previsoes).mean(axis=0).tolist()
|
||||
classes = self._classes_por_tarefa[tarefa]
|
||||
if len(valores) != len(classes):
|
||||
raise RuntimeError(
|
||||
f"Quantidade de previsões incompatível para {tarefa}: "
|
||||
f"{len(valores)} valores e {len(classes)} classes."
|
||||
)
|
||||
pontuacoes = (
|
||||
PontuacaoDeEtiqueta(etiqueta=etiqueta, confianca=float(valor))
|
||||
for etiqueta, valor in zip(classes, valores)
|
||||
)
|
||||
return tuple(sorted(pontuacoes, key=lambda item: item.confianca, reverse=True))
|
||||
@@ -10,6 +10,14 @@ from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PontuacaoDeEtiqueta:
|
||||
"""Representa uma etiqueta musical e a confiança calculada para ela."""
|
||||
|
||||
etiqueta: str
|
||||
confianca: float
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ResultadoDaAnaliseMusical:
|
||||
"""
|
||||
@@ -31,5 +39,7 @@ class ResultadoDaAnaliseMusical:
|
||||
humores: tuple[str, ...] = field(default_factory=tuple)
|
||||
instrumentos: tuple[str, ...] = field(default_factory=tuple)
|
||||
etiquetas: tuple[str, ...] = field(default_factory=tuple)
|
||||
pontuacoes_de_genero: tuple[PontuacaoDeEtiqueta, ...] = field(default_factory=tuple)
|
||||
pontuacoes_de_humor: tuple[PontuacaoDeEtiqueta, ...] = field(default_factory=tuple)
|
||||
provedor: str = "Essentia"
|
||||
modelo: str | None = None
|
||||
|
||||
@@ -26,10 +26,12 @@ class AnalisadorDeMusicaInstrumentalEssentia:
|
||||
self,
|
||||
extrator: Callable[[str], tuple[Any, Any]] | None = None,
|
||||
nome_do_modelo: str | None = None,
|
||||
classificador: Any | None = None,
|
||||
) -> None:
|
||||
"""Inicializa o adaptador sem carregar a dependência externa."""
|
||||
self._extrator = extrator
|
||||
self._nome_do_modelo = nome_do_modelo
|
||||
self._classificador = classificador
|
||||
|
||||
def analisar(self, arquivo: str | Path) -> ResultadoDaAnaliseMusical:
|
||||
"""
|
||||
@@ -53,7 +55,13 @@ class AnalisadorDeMusicaInstrumentalEssentia:
|
||||
dados, _ = self._obter_extrator()(str(caminho))
|
||||
if dados is None:
|
||||
raise ValueError("O Essentia devolveu uma análise vazia.")
|
||||
return self._converter_resultado(caminho, dados)
|
||||
pontuacoes_de_genero: tuple[Any, ...] = ()
|
||||
pontuacoes_de_humor: tuple[Any, ...] = ()
|
||||
if self._classificador is not None:
|
||||
pontuacoes_de_genero, pontuacoes_de_humor = self._classificador.classificar(caminho)
|
||||
return self._converter_resultado(
|
||||
caminho, dados, pontuacoes_de_genero, pontuacoes_de_humor,
|
||||
)
|
||||
|
||||
def _obter_extrator(self) -> Callable[[str], tuple[Any, Any]]:
|
||||
if self._extrator is not None:
|
||||
@@ -69,7 +77,11 @@ class AnalisadorDeMusicaInstrumentalEssentia:
|
||||
return self._extrator
|
||||
|
||||
def _converter_resultado(
|
||||
self, caminho: Path, dados: Any,
|
||||
self,
|
||||
caminho: Path,
|
||||
dados: Any,
|
||||
pontuacoes_de_genero: tuple[Any, ...] = (),
|
||||
pontuacoes_de_humor: tuple[Any, ...] = (),
|
||||
) -> ResultadoDaAnaliseMusical:
|
||||
return ResultadoDaAnaliseMusical(
|
||||
caminho_do_arquivo=caminho,
|
||||
@@ -79,10 +91,12 @@ class AnalisadorDeMusicaInstrumentalEssentia:
|
||||
modo=self._texto(self._buscar(dados, "tonal.key_edma.scale")),
|
||||
intensidade=self._numero(self._buscar(dados, "lowlevel.average_loudness")),
|
||||
dancabilidade=self._numero(self._buscar(dados, "rhythm.danceability")),
|
||||
generos=self._etiquetas(dados, ("genre", "genres")),
|
||||
humores=self._etiquetas(dados, ("mood", "moods")),
|
||||
generos=tuple(item.etiqueta for item in pontuacoes_de_genero),
|
||||
humores=tuple(item.etiqueta for item in pontuacoes_de_humor),
|
||||
instrumentos=self._etiquetas(dados, ("instrument", "instruments")),
|
||||
etiquetas=self._etiquetas(dados, ("tags", "semantic")),
|
||||
pontuacoes_de_genero=pontuacoes_de_genero,
|
||||
pontuacoes_de_humor=pontuacoes_de_humor,
|
||||
modelo=self._nome_do_modelo,
|
||||
)
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@ import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from engine.integracoes.audio import AnalisadorDeMusicaInstrumentalEssentia
|
||||
from engine.integracoes.audio import PontuacaoDeEtiqueta
|
||||
|
||||
|
||||
class TesteAnalisadorDeMusicaInstrumentalEssentia(unittest.TestCase):
|
||||
@@ -54,3 +55,25 @@ class TesteAnalisadorDeMusicaInstrumentalEssentia(unittest.TestCase):
|
||||
self.assertEqual(resultado.duracao_em_segundos, 10.0)
|
||||
self.assertEqual(resultado.batidas_por_minuto, 110.0)
|
||||
self.assertEqual(resultado.tonalidade, "D")
|
||||
|
||||
def test_incorpora_classificacoes_do_audio_ao_resultado(self):
|
||||
classificador = type("Classificador", (), {
|
||||
"classificar": lambda self, _arquivo: (
|
||||
(PontuacaoDeEtiqueta("Jazz-Funk", 0.8),),
|
||||
(PontuacaoDeEtiqueta("energética", 0.7),),
|
||||
),
|
||||
})()
|
||||
|
||||
def extrator(_caminho: str):
|
||||
return {"rhythm.bpm": 100.0}, {}
|
||||
|
||||
with tempfile.TemporaryDirectory() as pasta:
|
||||
arquivo = Path(pasta) / "instrumental.wav"
|
||||
arquivo.write_bytes(b"audio")
|
||||
resultado = AnalisadorDeMusicaInstrumentalEssentia(
|
||||
extrator, classificador=classificador,
|
||||
).analisar(arquivo)
|
||||
|
||||
self.assertEqual(resultado.generos, ("Jazz-Funk",))
|
||||
self.assertEqual(resultado.humores, ("energética",))
|
||||
self.assertEqual(resultado.pontuacoes_de_genero[0].confianca, 0.8)
|
||||
|
||||
Reference in New Issue
Block a user