feat: adicionadas métricas opcionais de fala ao Scanner com anális
- 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
This commit is contained in:
@@ -0,0 +1,5 @@
|
||||
"""Integrações locais para análise de áudio."""
|
||||
|
||||
from .analisador_de_metricas_de_voz import AnalisadorDeMetricasDeVoz
|
||||
|
||||
__all__ = ["AnalisadorDeMetricasDeVoz"]
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user