From b03b17597303c1dcd253571082af0b7ff01d1526 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Henrique?= Date: Wed, 9 Sep 2026 17:17:55 -0400 Subject: [PATCH] =?UTF-8?q?feat:=20adicionadas=20m=C3=A9tricas=20opcionais?= =?UTF-8?q?=20de=20fala=20ao=20Scanner=20com=20an=C3=A1lis?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - adicionadas métricas opcionais de fala ao Scanner com análise acústica na engine Resumo: - 7 arquivos alterados - 2 novos - 5 modificados - 0 removidos 5 files changed, 43 insertions(+), 6 deletions(-) Arquivos: - code/cep-plugin/index.html - code/cep-plugin/main.js - code/engine/executar_scanner.py - code/engine/scanner/configuracao_visual.py - code/engine/scanner/transcricao_da_timeline.py - code/engine/integracoes/audio/ - code/engine/testes/test_analisador_de_metricas_de_voz.py --- code/cep-plugin/index.html | 2 + code/cep-plugin/main.js | 3 +- code/engine/executar_scanner.py | 14 +- code/engine/integracoes/audio/__init__.py | 5 + .../audio/analisador_de_metricas_de_voz.py | 141 ++++++++++++++++++ code/engine/scanner/configuracao_visual.py | 3 + .../engine/scanner/transcricao_da_timeline.py | 27 +++- .../test_analisador_de_metricas_de_voz.py | 52 +++++++ 8 files changed, 241 insertions(+), 6 deletions(-) create mode 100644 code/engine/integracoes/audio/__init__.py create mode 100644 code/engine/integracoes/audio/analisador_de_metricas_de_voz.py create mode 100644 code/engine/testes/test_analisador_de_metricas_de_voz.py diff --git a/code/cep-plugin/index.html b/code/cep-plugin/index.html index 292403f..21a4cbc 100755 --- a/code/cep-plugin/index.html +++ b/code/cep-plugin/index.html @@ -315,6 +315,8 @@

Transcrição

—
+ +

Mede pitch, energia, velocidade de fala e pausas por trecho. Não classifica emoção e tem custo baixo comparado à transcrição.

Hugging Face não configurado

A lista inclui somente modelos encontrados nas pastas locais configuradas. O token do Hugging Face é usado apenas na diarização; nunca é salvo no perfil nem no relatório.

diff --git a/code/cep-plugin/main.js b/code/cep-plugin/main.js index 7afe55e..5d830ab 100755 --- a/code/cep-plugin/main.js +++ b/code/cep-plugin/main.js @@ -1159,7 +1159,7 @@ function scannerProfileFromScreen() { include("faces", ["qualidade_facial"]); include("pose", ["maos"]); include("ocr", ["codigos"]); include("estetica", ["horizonte", "retangulos", "contornos"]); var modelo = document.getElementById("scannerTranscribeModel").value; var opcao = document.getElementById("scannerTranscribeModel").selectedOptions[0]; - return { versao: 1, modo_de_analise: scannerModoAtual(), intervalo_em_segundos: Number(document.getElementById("scannerInterval").value), resolucao_maxima: Number(document.getElementById("scannerResolution").value), faixas_de_video: scannerSelectedTracks("video"), faixas_de_audio: scannerSelectedTracks("audio"), recursos_vision: SCANNER_ALL_RESOURCES.filter(function (item) { return enabled.indexOf(item) >= 0; }), detectar_cenas: document.getElementById("scannerScenes").checked, analisar_continuidade: document.getElementById("scannerContinuity").checked, preparar_reenquadramento: document.getElementById("scannerReframe").checked, interpretar_cena: document.getElementById("scannerInterpret").checked, transcricao: { modelo: modelo, caminho_modelo: opcao && opcao.dataset.path || "", idioma: document.getElementById("scannerTranscribeLanguage").value, diarizacao: document.getElementById("scannerDiarization").checked, usar_hugging_face: !!loadHfToken() } }; + return { versao: 1, modo_de_analise: scannerModoAtual(), metricas_de_fala: document.getElementById("scannerVoiceFeatures").checked, intervalo_em_segundos: Number(document.getElementById("scannerInterval").value), resolucao_maxima: Number(document.getElementById("scannerResolution").value), faixas_de_video: scannerSelectedTracks("video"), faixas_de_audio: scannerSelectedTracks("audio"), recursos_vision: SCANNER_ALL_RESOURCES.filter(function (item) { return enabled.indexOf(item) >= 0; }), detectar_cenas: document.getElementById("scannerScenes").checked, analisar_continuidade: document.getElementById("scannerContinuity").checked, preparar_reenquadramento: document.getElementById("scannerReframe").checked, interpretar_cena: document.getElementById("scannerInterpret").checked, transcricao: { modelo: modelo, caminho_modelo: opcao && opcao.dataset.path || "", idioma: document.getElementById("scannerTranscribeLanguage").value, diarizacao: document.getElementById("scannerDiarization").checked, usar_hugging_face: !!loadHfToken() } }; } function saveScannerProfile() { @@ -1179,6 +1179,7 @@ function loadScannerProfile() { var resources = saved.recursos_vision || SCANNER_ALL_RESOURCES; Array.prototype.slice.call(document.querySelectorAll("[data-scanner-resource]")).forEach(function (item) { item.checked = resources.indexOf(item.getAttribute("data-scanner-resource")) >= 0; }); document.getElementById("scannerScenes").checked = saved.detectar_cenas !== false; document.getElementById("scannerContinuity").checked = saved.analisar_continuidade !== false; document.getElementById("scannerReframe").checked = !!saved.preparar_reenquadramento; document.getElementById("scannerInterpret").checked = saved.interpretar_cena !== false; + document.getElementById("scannerVoiceFeatures").checked = !!saved.metricas_de_fala; if (saved.transcricao) { document.getElementById("scannerTranscribeModel").value = saved.transcricao.modelo || ""; document.getElementById("scannerTranscribeLanguage").value = saved.transcricao.idioma || loadLanguage(); document.getElementById("scannerDiarization").checked = !!saved.transcricao.diarizacao && !!loadHfToken(); } scannerAtualizarModo(); } diff --git a/code/engine/executar_scanner.py b/code/engine/executar_scanner.py index dcaead0..7a4bf12 100644 --- a/code/engine/executar_scanner.py +++ b/code/engine/executar_scanner.py @@ -133,13 +133,21 @@ def executar(entrada: Path, saida: Path) -> Path: audio_arquivos = [] transcricoes_por_clipe: dict[str, list[dict]] = {} transcricao_configurada = pedido["perfil"].get("transcricao", {}) + transcricao_configurada = { + **transcricao_configurada, + "metricas_de_fala": configuracao.metricas_de_fala, + } if faixas_de_audio and transcricao_configurada.get("caminho_modelo"): try: from engine.integracoes.whisper import ProviderDeTranscricaoLocal + from engine.integracoes.audio import AnalisadorDeMetricasDeVoz from engine.dominio import Timeline emitir("Transcrevendo áudio", 5) provider = ProviderDeTranscricaoLocal(transcricao_configurada["caminho_modelo"], idioma=transcricao_configurada.get("idioma", "pt")) + analisador_de_metricas_de_voz = ( + AnalisadorDeMetricasDeVoz() if configuracao.metricas_de_fala else None + ) diarizador = None if transcricao_configurada.get("diarizacao"): if not os.environ.get("HF_TOKEN"): @@ -152,14 +160,16 @@ def executar(entrada: Path, saida: Path) -> Path: transcricao_configurada = {**transcricao_configurada, "modelo_diarizacao": modelo_diarizacao} transcricoes = TranscricaoDaTimeline(provider, diretoria_de_trabalho=pasta / ".cache-audio", - diarizador=diarizador).executar( + diarizador=diarizador, + analisador_de_metricas_de_voz=analisador_de_metricas_de_voz).executar( Timeline(timeline.identificador, timeline.nome, faixas=faixas_de_audio)) # Persiste a transcrição (segmentos + falas) no banco de análises. repositorio_de_analises.registrar_transcricoes(timeline.identificador, transcricoes) for transcricao in transcricoes: transcricoes_por_clipe[transcricao.identificador_do_clipe] = [ {"inicio": item.inicio, "fim": item.fim, "texto": item.texto, - "confianca": item.confianca, "falante": item.falante} + "confianca": item.confianca, "falante": item.falante, + "caracteristicas_acusticas": item.caracteristicas_acusticas} for item in transcricao.segmentos] except Exception as erro: transcricao_configurada = {**transcricao_configurada, "erro": str(erro)} diff --git a/code/engine/integracoes/audio/__init__.py b/code/engine/integracoes/audio/__init__.py new file mode 100644 index 0000000..d080080 --- /dev/null +++ b/code/engine/integracoes/audio/__init__.py @@ -0,0 +1,5 @@ +"""Integrações locais para análise de áudio.""" + +from .analisador_de_metricas_de_voz import AnalisadorDeMetricasDeVoz + +__all__ = ["AnalisadorDeMetricasDeVoz"] diff --git a/code/engine/integracoes/audio/analisador_de_metricas_de_voz.py b/code/engine/integracoes/audio/analisador_de_metricas_de_voz.py new file mode 100644 index 0000000..6790573 --- /dev/null +++ b/code/engine/integracoes/audio/analisador_de_metricas_de_voz.py @@ -0,0 +1,141 @@ +"""Calcula sinais acústicos determinísticos para trechos transcritos. + +Este módulo concentra a lógica usada pela tela Editar vídeo para que o +Scanner também possa calcular as mesmas métricas sem depender de um script da +interface. As métricas são sinais de apoio para uma análise posterior; elas +não classificam emoção sozinhas. +""" + +from pathlib import Path +from typing import Any, Sequence +import wave + + +class AnalisadorDeMetricasDeVoz: + """Calcula energia, pitch, velocidade e pausas em arquivos WAV.""" + + TAXA_DE_AMOSTRAGEM = 16000 + DURACAO_DA_JANELA_MS = 30 + PASSO_DA_JANELA_MS = 10 + PITCH_MINIMO_HZ = 75 + PITCH_MAXIMO_HZ = 400 + + def __init__(self) -> None: + """Inicializa o analisador sem carregar dependências pesadas.""" + self._amostras_por_arquivo: dict[str, Any] = {} + + def analisar( + self, + arquivo: str | Path, + inicio: float, + fim: float, + palavras: Sequence[Any] = (), + inicio_absoluto: float | None = None, + fim_anterior: float | None = None, + ) -> dict[str, Any]: + """Calcula as métricas de um trecho de WAV já normalizado.""" + amostras = self._carregar_amostras(arquivo) + metricas = self._calcular_pitch_e_energia(amostras, inicio, fim) + duracao = fim - inicio + metricas["speaking_rate_wps"] = len(palavras) / duracao if duracao > 0 else None + metricas["longest_internal_pause_s"] = self._maior_pausa_interna(palavras) + metricas["gap_before_s"] = ( + inicio_absoluto - fim_anterior + if inicio_absoluto is not None and fim_anterior is not None + else None + ) + return metricas + + def _carregar_amostras(self, arquivo: str | Path) -> Any: + caminho = str(Path(arquivo).expanduser().resolve()) + if caminho not in self._amostras_por_arquivo: + try: + import numpy as np + except ImportError as erro: + raise RuntimeError( + "As métricas de voz precisam da biblioteca numpy." + ) from erro + with wave.open(caminho, "rb") as arquivo_wav: + taxa = arquivo_wav.getframerate() + quantidade_de_canais = arquivo_wav.getnchannels() + largura_da_amostra = arquivo_wav.getsampwidth() + dados = arquivo_wav.readframes(arquivo_wav.getnframes()) + tipos_por_largura = {1: np.uint8, 2: np.int16, 4: np.int32} + tipo_da_amostra = tipos_por_largura.get(largura_da_amostra) + if tipo_da_amostra is None: + raise ValueError(f"Largura de amostra WAV não suportada: {largura_da_amostra} bytes.") + amostras = np.frombuffer(dados, dtype=tipo_da_amostra) + if largura_da_amostra == 1: + amostras = (amostras.astype(np.float32) - 128.0) / 128.0 + else: + amostras = amostras.astype(np.float32) / np.iinfo(tipo_da_amostra).max + if quantidade_de_canais > 1: + amostras = amostras.reshape(-1, quantidade_de_canais).mean(axis=1) + if taxa != self.TAXA_DE_AMOSTRAGEM: + raise ValueError( + f"Taxa de amostragem inesperada: {taxa}; " + f"esperado {self.TAXA_DE_AMOSTRAGEM}." + ) + self._amostras_por_arquivo[caminho] = amostras + return self._amostras_por_arquivo[caminho] + + def _calcular_pitch_e_energia(self, amostras: Any, inicio: float, fim: float) -> dict[str, Any]: + import numpy as np + + inicio_amostra = max(0, int(inicio * self.TAXA_DE_AMOSTRAGEM)) + fim_amostra = min(len(amostras), int(fim * self.TAXA_DE_AMOSTRAGEM)) + trecho = amostras[inicio_amostra:fim_amostra] + energia = float(np.sqrt(np.mean(np.square(trecho)))) if trecho.size else 0.0 + tamanho_da_janela = int(self.TAXA_DE_AMOSTRAGEM * self.DURACAO_DA_JANELA_MS / 1000) + passo_da_janela = int(self.TAXA_DE_AMOSTRAGEM * self.PASSO_DA_JANELA_MS / 1000) + extensao = max(0, len(trecho) - tamanho_da_janela) + quantidade_de_janelas = -(-extensao // passo_da_janela) if extensao else 0 + if quantidade_de_janelas: + janelas = np.lib.stride_tricks.sliding_window_view( + trecho, tamanho_da_janela + )[::passo_da_janela][:quantidade_de_janelas] + pitches = self._estimar_pitch_em_lote(janelas) + pitches = pitches[~np.isnan(pitches)] + else: + pitches = np.array([], dtype=np.float64) + return { + "energy_rms": energia, + "pitch_hz_median": float(np.median(pitches)) if pitches.size else None, + "pitch_hz_std": float(np.std(pitches)) if pitches.size else None, + } + + def _estimar_pitch_em_lote(self, janelas: Any) -> Any: + import numpy as np + + quantidade_de_janelas, tamanho_da_janela = janelas.shape + resultado = np.full(quantidade_de_janelas, np.nan, dtype=np.float64) + if not quantidade_de_janelas: + return resultado + energia = np.sqrt(np.mean(np.square(janelas), axis=1)) + centralizadas = janelas - janelas.mean(axis=1, keepdims=True) + tamanho_fft = 1 + while tamanho_fft < 2 * tamanho_da_janela: + tamanho_fft *= 2 + espectro = np.fft.rfft(centralizadas, n=tamanho_fft, axis=1) + potencia = espectro.real ** 2 + espectro.imag ** 2 + autocorrelacao = np.fft.irfft(potencia, n=tamanho_fft, axis=1)[:, :tamanho_da_janela] + atraso_minimo = int(self.TAXA_DE_AMOSTRAGEM / self.PITCH_MAXIMO_HZ) + atraso_maximo = int(self.TAXA_DE_AMOSTRAGEM / self.PITCH_MINIMO_HZ) + janela = autocorrelacao[:, atraso_minimo:atraso_maximo] + indice_do_pico = np.argmax(janela, axis=1) + valor_do_pico = janela[np.arange(quantidade_de_janelas), indice_do_pico] + atraso_do_pico = indice_do_pico + atraso_minimo + with np.errstate(divide="ignore", invalid="ignore"): + confianca = np.where(autocorrelacao[:, 0] > 0, + valor_do_pico / autocorrelacao[:, 0], 0.0) + validos = (energia >= 1e-4) & (autocorrelacao[:, 0] > 0) & (confianca >= 0.3) + resultado[validos] = self.TAXA_DE_AMOSTRAGEM / atraso_do_pico[validos] + return resultado + + @staticmethod + def _maior_pausa_interna(palavras: Sequence[Any]) -> float: + """Obtém a maior pausa entre duas palavras do mesmo segmento.""" + maior_pausa = 0.0 + for anterior, atual in zip(palavras, palavras[1:]): + maior_pausa = max(maior_pausa, atual.inicio - anterior.fim) + return maior_pausa diff --git a/code/engine/scanner/configuracao_visual.py b/code/engine/scanner/configuracao_visual.py index 2f76233..9c9e425 100644 --- a/code/engine/scanner/configuracao_visual.py +++ b/code/engine/scanner/configuracao_visual.py @@ -39,6 +39,7 @@ class ConfiguracaoVisualDoScanner: faixas_de_video: tuple[int, ...] = () faixas_de_audio: tuple[int, ...] = () modo_de_analise: str = "ambos" + metricas_de_fala: bool = False recursos_vision: frozenset[str] = RECURSOS_VISION detectar_cenas: bool = True analisar_continuidade: bool = True @@ -87,6 +88,7 @@ class ConfiguracaoVisualDoScanner: faixas_de_video=tuple(sorted(set(int(item) for item in dados.get("faixas_de_video", ())))), faixas_de_audio=tuple(sorted(set(int(item) for item in dados.get("faixas_de_audio", ())))), modo_de_analise=str(dados.get("modo_de_analise", "ambos")), + metricas_de_fala=bool(dados.get("metricas_de_fala", False)), recursos_vision=frozenset(dados.get("recursos_vision", RECURSOS_VISION)), detectar_cenas=bool(dados.get("detectar_cenas", True)), analisar_continuidade=bool(dados.get("analisar_continuidade", True)), @@ -108,6 +110,7 @@ class ConfiguracaoVisualDoScanner: "faixas_de_video": list(self.faixas_de_video), "faixas_de_audio": list(self.faixas_de_audio), "modo_de_analise": self.modo_de_analise, + "metricas_de_fala": self.metricas_de_fala, "recursos_vision": sorted(self.recursos_vision), "detectar_cenas": self.detectar_cenas, "analisar_continuidade": self.analisar_continuidade, diff --git a/code/engine/scanner/transcricao_da_timeline.py b/code/engine/scanner/transcricao_da_timeline.py index b66da84..035f2ff 100644 --- a/code/engine/scanner/transcricao_da_timeline.py +++ b/code/engine/scanner/transcricao_da_timeline.py @@ -34,12 +34,14 @@ class TranscricaoDaTimeline: def __init__(self, provider: Any, extrator: ExtracaoDeAudio | None = None, diretoria_de_trabalho: str | Path = ".transcricao", analisador_de_emocao: Any | None = None, - diarizador: Any | None = None) -> None: + diarizador: Any | None = None, + analisador_de_metricas_de_voz: Any | None = None) -> None: self.provider = provider self.extrator = extrator or ExtracaoDeAudio() self.diretoria_de_trabalho = Path(diretoria_de_trabalho) self.analisador_de_emocao = analisador_de_emocao self.diarizador = diarizador + self.analisador_de_metricas_de_voz = analisador_de_metricas_de_voz def executar(self, timeline: Any) -> list[TranscricaoDoClipe]: clipes: list[Any] = [] @@ -113,15 +115,23 @@ class TranscricaoDaTimeline: dados = json.loads(cache.read_text(encoding="utf-8")) if all("palavras" in item and (self.analisador_de_emocao is None or - ("emocao" in item and "voz_aparente" in item)) + ("emocao" in item and "voz_aparente" in item)) and + (self.analisador_de_metricas_de_voz is None or + "caracteristicas_acusticas" in item) for item in dados): return [self._segmento_do_json(item) for item in dados] partes = self.extrator.extrair(arquivo, pasta / "audio") segmentos: list[SegmentoDeTranscricao] = [] + fim_anterior: float | None = None for parte in partes: falas = self._diarizar(parte.caminho) for segmento in self.provider.transcrever(type("Audio", (), {"arquivo": str(parte.caminho)})()): analise = self._analisar_emocao(parte.caminho, segmento.inicio, segmento.fim) + inicio_absoluto = parte.inicio + segmento.inicio + metricas_de_voz = self._analisar_metricas_de_voz( + parte.caminho, segmento.inicio, segmento.fim, segmento.palavras, + inicio_absoluto, fim_anterior, + ) falante = self._falante_em(falas, segmento.inicio, segmento.fim) segmentos.append(SegmentoDeTranscricao(parte.inicio + segmento.inicio, parte.inicio + segmento.fim, segmento.texto, segmento.confianca, @@ -130,8 +140,9 @@ class TranscricaoDaTimeline: self._falante_em(falas, p.inicio, p.fim)) for p in segmento.palavras), analise.get("emocao"), analise.get("confianca"), - dict(analise.get("caracteristicas_acusticas", {})), falante, + metricas_de_voz or dict(analise.get("caracteristicas_acusticas", {})), falante, analise.get("voz_aparente"), analise.get("confianca_voz"))) + fim_anterior = inicio_absoluto + segmento.fim - segmento.inicio pasta.mkdir(parents=True, exist_ok=True) cache.write_text(json.dumps([asdict(item) for item in segmentos], ensure_ascii=False, indent=2), encoding="utf-8") (pasta / f"{prefixo}.txt").write_text( @@ -157,6 +168,16 @@ class TranscricaoDaTimeline: return {} return dict(self.analisador_de_emocao.analisar(arquivo, inicio, fim)) + def _analisar_metricas_de_voz(self, arquivo: Path, inicio: float, fim: float, + palavras: tuple[PalavraDeTranscricao, ...], + inicio_absoluto: float, + fim_anterior: float | None) -> dict[str, Any]: + """Calcula métricas acústicas somente quando foram solicitadas.""" + if self.analisador_de_metricas_de_voz is None: + return {} + return dict(self.analisador_de_metricas_de_voz.analisar( + arquivo, inicio, fim, palavras, inicio_absoluto, fim_anterior)) + def _diarizar(self, arquivo: Path) -> list[tuple[float, float, str]]: if self.diarizador is None: return [] diff --git a/code/engine/testes/test_analisador_de_metricas_de_voz.py b/code/engine/testes/test_analisador_de_metricas_de_voz.py new file mode 100644 index 0000000..1c09d30 --- /dev/null +++ b/code/engine/testes/test_analisador_de_metricas_de_voz.py @@ -0,0 +1,52 @@ +"""Testes do cálculo de métricas acústicas do Scanner.""" + +import math +import tempfile +import unittest +import wave +from pathlib import Path +from types import SimpleNamespace + +from engine.integracoes.audio import AnalisadorDeMetricasDeVoz + + +class TesteDoAnalisadorDeMetricasDeVoz(unittest.TestCase): + """Verifica a interface pública do analisador de voz.""" + + def test_calcula_metricas_do_trecho_e_reaproveita_o_audio(self): + """Calcula sinais básicos e mantém a pausa entre palavras.""" + with tempfile.TemporaryDirectory() as pasta: + arquivo = Path(pasta) / "fala.wav" + self._criar_wav(arquivo) + palavras = ( + SimpleNamespace(inicio=0.0, fim=0.25), + SimpleNamespace(inicio=0.5, fim=0.75), + ) + analisador = AnalisadorDeMetricasDeVoz() + primeira = analisador.analisar(arquivo, 0.0, 1.0, palavras) + segunda = analisador.analisar(arquivo, 1.0, 2.0, (), 1.5, 1.0) + + self.assertGreater(primeira["energy_rms"], 0) + self.assertIsNotNone(primeira["pitch_hz_median"]) + self.assertAlmostEqual(primeira["speaking_rate_wps"], 2.0) + self.assertAlmostEqual(primeira["longest_internal_pause_s"], 0.25) + self.assertAlmostEqual(segunda["gap_before_s"], 0.5) + + @staticmethod + def _criar_wav(caminho: Path) -> None: + """Cria um WAV mono curto e determinístico para o teste.""" + taxa = 16000 + amostras = [ + int(12000 * math.sin(2 * math.pi * 220 * indice / taxa)) + for indice in range(taxa * 2) + ] + with wave.open(str(caminho), "wb") as arquivo: + arquivo.setnchannels(1) + arquivo.setsampwidth(2) + arquivo.setframerate(taxa) + arquivo.writeframes(b"".join(int(amostra).to_bytes(2, "little", signed=True) + for amostra in amostras)) + + +if __name__ == "__main__": + unittest.main()