Este programa foi marcado como "Limpo" pelo anfitrião.
Python e suas bibliotecas
1. librosa — carregamento e ressamplo de áudio
def carregar_áudio(caminho):
y, sr = librosa.carregar(caminho, sr=SR, mono=True)
return y, sr
Em cada episódio, independentemente da taxa de amostragem original ou do número de canais, ele é carregado para um único fluxo mono a 16kHz (SR = 16000). Essa é uma linha de código do librosa, reamostragem e conversão estéreo para mono. Isso fez com que todas as ferramentas de baixo (Detecção de Perda de Faleza, Acompanhamento de Tomada de Som, Extração de MFCC) operassem em uma entrada consistente e previsível, independentemente do que um codificador MP3 ou um anfitrião de podcast tenha usado originalmente.
2. parselmouth (ligações Praat) — rastreamento de tom e qualidade da voz
def pitch_e_hnr(y, sr):
snd = parselmouth.Sound(y, frequência_amostragem=sr)
pitch = snd.para_pitch(passo_tempo=PITCH_TIME_STEP, piso_pitch=PITCH_FLOOR, teto_pitch=PITCH_CEILING)
f0 = pitch.array_selecionado["frequência"]
tempos = pitch.xs()
try:
harmonicidade = snd.para_harmonicidade_cc(passo_tempo=PITCH_TIME_STEP)
hnr = np.array([harmonicidade.get_valor(t) for t in tempos])
hnr = np.nan_to_num(hnr, nan=-999.0)
except Exception:
hnr = np.full_like(f0, 999.0) # falha aberta se houver erros na extração de HNR
return f0, hnr, tempos
Parselmouth é um wrapper Python para o Praat, a ferramenta padrão em pesquisa fonética. Este é o núcleo de toda a medição de monotonia vocal: `to_pitch()` fornece a frequência fundamental (F0, aproximadamente "timbre") a cada 20ms, e `to_harmonicity_cc()` dá a razão harmônicos-para-ruído (HNR) nos mesmos marcadores de tempo — uma pontuação de qualidade do sinal usada posteriormente para descartar estimativas de timbre extraídas de áudios ruidosos ou mal comprimidos, em vez de permitir que áudios ruins se passem por "timbre plano".
3. webrtcvad — decidindo quais quadros são realmente fala
def vad_speech_mask(y, sr, agressividade=2):
vad = webrtcvad.Vad(agressividade)
pcm16 = (np.clip(y, -1.0, 1.0) * 32767).astype(np.int16)
n_frames = len(pcm16) // VAD_FRAME_LEN
mascara = np.zeros(n_frames, dtype=bool)
for i in range(n_frames):
chunk = pcm16[i * VAD_FRAME_LEN: (i + 1) * VAD_FRAME_LEN]
raw = struct.pack("%dh" % len(chunk), *chunk)
try:
mascara[i] = vad.is_speech(raw, sr)
except Exception:
mascara[i] = False
return mascara, tempos
O webrtcvad é o detector de atividade vocal do projeto WebRTC do Google — um classificador frame-a-frame ("esta parcela de 30ms é fala ou não"). Ele requer bytes PCM 16-bit cru, não floats, portanto a conversão com np.clip e struct.pack.
4. librosa.feature — diferenciando música da fala
flatness = librosa.feature.spectral_flatness(y=y, hop_length=hop)[0] chroma = librosa.feature.chroma_stft(y=y, sr=sr, hop_length=hop) onset_env = librosa.onset.onset_strength(y=y, sr=sr, hop_length=hop) ... chroma_diff = np.mean(np.linalg.norm(np.diff(chroma_win, axis=1), axis=0)) chroma_stability = 1.0 / (1.0 + chroma_diff) ... scores[w] = 0.4 * chroma_stability + 0.35 * periodicity + 0.25 * min(flat_mean * 3, 1.0)
Intros e outros de podcasts são frequentemente acompanhados de música, e a música tem um perfil de tom muito diferente da fala — incluindo isso poderia influenciar a medição de "monotonia". Em vez de treinar um classificador, três características do Librosa foram combinadas com pesos ajustados manualmente: planicidade espectral (a música tende a ter uma estrutura tonal mais marcada e um espectro menos ruidoso) e estabilidade de cromatismo (harmonia sustentada versus
(formantes da fala em constante mudança) e a periodicidade de início (um ritmo regular versus o ritmo irregular da fala).
5. librosa.feature.mfcc + sklearn.cluster.KMeans + sklearn.metrics.silhouette_score — uma tentativa não supervisionada de contagem de oradores
`mfcc = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=13, hop_length=hop)`
Os coeficientes cepstrais de frequência mel (MFCCs) resumem a forma do espectro de uma janela de áudio curta, um 'impressão digital de voz' amplamente utilizado em tarefas de fala. A ideia é dividir cada episódio em pedaços de 2 segundos, transformar cada um em um vetor de impressão digital, padronizar (valor z) as características, e usar KMeans para ver se os pedaços se dividem naturalmente em 2+ clusters (diferentes vozes) em vez de 1 (uma única voz durante todo o período).
6. scipy.stats.mannwhitneyu e uma mão escrita Cliff's delta — comparando dois grupos sem assumir uma distribuição normal.
`def cliffs_delta(a, b):`
scipy.stats.mannwhitneyu é um teste estatístico: ele compara dois grupos de números (HPR vs. podcasts de comparação, por métrica) sem assumir que eles estão distribuídos normalmente — apropriado para dados do mundo real assimétricos, como medições de alcance de tom. O scipy não vem com uma medida de efeito junto, então Cliff's delta — 'de todas as possíveis (episódio HPR, episódio de comparação) pares, com que frequência o valor do HPR é o menor?' Este par (um teste de significância mais um número de efeito) foi usado durante todo o processo porque um valor p baixo sozinho não diz se a diferença é realmente grande o suficiente para importar.
7. pandas — agrupamento, estratificação e tabulação cruzada de resultados.
`print(df.groupby("corpus")["retained_fraction"].describe())`
Uma vez que os dados/resultados.jsonl (um objeto JSON por episódio analisado) é carregado em um DataFrame do pandas, duas linhas de código respondem a duas perguntas diferentes que, de outra forma, exigiriam código personalizado: .groupby(...) .describe() fornece um resumo da distribuição (contagem, média, desvio padrão, quartis) da cobertura de qualidade de áudio por corpus, em uma chamada, e pd.crosstab() tabula HPR vs. comparação contra registro típico masculino/típico feminino — uma verificação rápida da sanidade que as duas coleções de podcasts tinham uma mistura semelhante de registro de gênero.
8. matplotlib — gráficos de caixa para comparação de manchetes
fig, ax = plt.subplots(figsize=(6, 4))
Um gráfico de caixa para cada métrica de alcance de pitch, HPR ao lado de podcasts de comparação, média e mediana marcadas.
9. vetores de palavras do spaCy — medindo o desvio de tema, não apenas a escolha de palavras
nlp = get_nlp() # spacy.load("en_core_web_md")
...
chunk_docs = [nlp(c) para c em chunks]
chunk_vecs = []
para d em chunk_docs:
v = d.vector
norm = np.linalg.norm(v)
se norm > 1e-6:
chunk_vecs.append(v / norm)
...
adj_sims = [float(np.dot(V[i], V[i + 1])) para i em range(n - 1)]
Cada pedaço de ~130 palavras de um transcrição é incorporado como um único vetor usando o modelo inglês médio do spaCy (en_core_web_md, que tem vetores de palavras embutidos), depois normalizado L2 para que um produto escalar entre dois vetoressejasua similaridade cosseno. Comparar a similaridade entre pedaços consecutivos dá uma pontuação de "o tema muda levemente de um pedaço para o próximo"?
riqueza léxica — diversidade de vocabulário que não engana na duração da transcrição.
lr = RiquezaLéxica(texto)
Uma pontuação ingênua de diversidade de vocabulário (palavras únicas ÷ palavras totais) piora à medida que o texto se estende, não porque o falante está realmente se repetindo mais, mas devido à definição da métrica.
O MTLD (Medida de Diversidade Léxica Textual) evita isso ao percorrer o texto e contar quantas palavras levam à queda da diversidade abaixo de um limiar, depois calculando a média — uma métrica projetada especificamente para permanecer comparável entre um episódio de 10 minutos e um de 40 minutos.
11. vaderSentiment — uma pontuação de sentimento baseada em léxico.
vader = obter_vader() # SentimentIntensityAnalyzer()
O VADER atribui um sentimento a cada pedaço de -1 a +1 usando um léxico e um conjunto de regras de mão (não uma rede neural) — escolhidos por serem rápidos, deterministas e baratos o suficiente para serem executados por pedaço em centenas de episódios. O valor usado não é a média do sentimento — é adesvio padrão.Analisar o sentimento em fragmentos de um episódio: o tom emocional varia ou permanece plano durante todo o episódio?
12. zlib (biblioteca padrão) — relação de compressão como indicador de entropia.
Definir relação de compressão(texto):
Sem bibliotecas de NLP de terceiros — apenas o módulo zlib da biblioteca padrão. Um compressor estilo gzip encontra e explora padrões repetidos; se conseguir reduzir muito um transcripção, ela era relativamente repetitiva e previsível, e se conseguir comprimi-la pouco, o texto era mais variado. Isso indica aproximadamente a "monotonia verbal".
13. re (biblioteca padrão) — transformando "HPR favorece apresentadores solo" em uma verificação heurística.
Padrões de múltiplos apresentadores:
Definir pontuação de múltiplos apresentadores(resumo, notas):
Em vez da tentativa de agrupamento de falantes acústicos (item 5, que provou ser pouco confiável), uma estatística mais confiável veio do texto simples: os metadados de episódios do HPR (extraídos de um repositório de base de conhecimento local) incluem resumos escritos pelo anfitrião, e uma lista curta de padrões de regex ("convidado", "entrevista", "se junta a mim", etc.) mais a contagem de perfis de correspondentes ligados foi suficiente para identificar episódios com múltiplos apresentadores.
14. Pool de multiprocessamento — executando a mesma análise em centenas de arquivos sem esperar centenas de vezes.
trabalho = [(r, caminhos_locais[r["id"]]) para r em lote] com mp.Pool(argumentos.trabalhadores) como pool, com arquivo aberto(argumentos.saida, "a") como arquivo_saida: para resultado em pool.imap_unordered(_trabalhador, trabalho): arquivo_saida.write(json.dumps(resultado) + "\n") arquivo_saida.flush()
Cada pipeline de lote (análise de tom acústico, passagem de contagem de oradores) utilizava a mesma estrutura: `Pool(workers).imap_unordered(...)` para executar a função por episódio em múltiplos núcleos de CPU simultaneamente, transmitindo os resultados para um arquivo JSON-lines à medida que cada um termina, em vez de esperar pela conclusão de todo o lote. `imap_unordered` foi escolhido em vez de `map` para que um único episódio lento não atrasasse a escrita dos resultados que já estavam prontos.
15. subprocesso + loop por lote resumível — trabalhando dentro de um limite de espaço em disco
`carregar_ids_concluidos(caminho_resultados):`
se `os.path.exists(caminho_resultados)` não for verdadeiro:
retornar conjunto vazio()
concluidos = conjunto()
com `open(caminho_resultados)` como f:
para linha em f:
...
concluidos.adicionar(json.loads(linha)["id"])
retornar concluidos
...
linhas = [r para r em linhas se r["id"] não estiver em concluidos]
...
para batch_start em intervalo(0, len(linhas), args.batch_size):
lote = linhas[batch_start:batch_start + args.batch_size]
caminhos_locais = buscar_lote(lote, args.scratch) # copiar do host remoto
...
para p em caminhos_locais.valores():
os.remove(p) # remover antes do próximo lote`
Todos os podcasts (~220GB) não cabiam no disco local (~54GB livres). A solução foi um loop: copiar (via subprocesso) uma leva de áudio, analisá-lo, excluí-lo, passar para a próxima leva. O `load_done_ids()` torna o pipeline resumível ao ler o que já está no arquivo de saída, então um crash ou uma parada deliberada a meio caminho não perde o material concluído.
Links
Hacker Public Radio: onde a comunidade tecnológica se reúne para compartilhar conhecimentos e criar um espaço único.
Visite o site: https://hackerpublicradio.org/
#HackerPublicRadio #ComunidadeTecnológica #Podcast #MagiaMonotona #ConteúdoFascinante #HackerNews #ngated