Reconhecimento de fala com Python e Vosk

Se você se interessa por reconhecimento de fala no Python, certamente já ouviu falar do pacote SpeechRecognition, que possibilita que você crie aplicações com reconhecimento de fala de um modo relativamente simples. Entretanto, esse pacote faz uso de recursos de terceiros para realizar o reconhecimento, muitos dos quais consistem em APIs online. A opção que temos para uso offline no Windows ou Linux é o CMUSphinx, um projeto que parece ter sido deixado de lado em prol do desenvolvimento de uma nova engine chamada Vosk.

No post de hoje, vou ensinar você a implementar reconhecimento de fala no Python, para conversão de fala em texto utilizando a Vosk. Assim, você vai poder rodar seu programa mesmo sem acesso à internet, com uma engine mais atual, que utiliza modelos relativamente leves. Inclusive, foi essa a biblioteca que utilizei para programar a Alda (minha assistente virtual sucessora do Geraldo).

Vamos desenvolver passo a passo dois pequenos scripts, que são versões simplificadas de scripts de exemplo da própria engine Vosk. O primeiro converte a fala contida em um arquivo de áudio do tipo wave em texto. O segundo captura o áudio do microfone e converte a fala detectada em texto em tempo real. No final do texto, também colocarei o código de um programinha que detecta a fala, converte para texto e depois fala o texto de novo com o pacote de síntese de voz que eu ensinei a usar nesse post aqui.

Os programinhas que criaremos aqui são bastante simples, mas abordaremos vários conceitos um tanto complexos, e entendendo bem o básico, você poderá fazer coisas muito divertidas, como um assistente virtual ou chatbot com síntese e reconhecimento de fala!

Instalando as bibliotecas necessárias

A biblioteca principal para realizar o reconhecimento de fala é a Vosk. Mas também precisaremos de uma biblioteca para captar o microfone e outra para carregar arquivos wave. Para a captura de áudio, utilizaremos o sounddevice, já para carregar os arquivos wave, utilizaremos a biblioteca wave.

Então, comecemos instalando as bibliotecas, caso você não tenha feito isso ainda. Todas elas podem ser instaladas facilmente através do pip. No Linux, basta executar os comandos:

pip3 install sounddevice wave vosk

No Linux, você deve ter o PortAudio instalado também. Para fazer isso (no Debian/Ubuntu), basta executar o comando:

sudo apt-get install portaudio19-dev 

No Windows, de acordo com os desenvolvedores da biblioteca, você pode baixar o wheel do sounddevice com a biblioteca PortAudio já embutida nesse link.

Finalmente, precisamos também de um modelo treinado para reconhecer palavras da língua portuguesa. Todo projeto que você utilizar com o Vosk deverá conter um diretório com o modelo a ser utilizado. Você pode obter um modelo do português clicando aqui. Caso você precise de um modelo para outra língua, existem vários disponíveis no site dos desenvolvedores do Vosk.

Em nosso exemplo, o diretório do modelo será chamada “modelo”. Então você pode criar uma pasta para o projeto e dentro dessa pasta, criar uma nova pasta chamada “modelo” para a qual você deverá extrair o conteúdo do arquivo .zip do modelo que você acabou de baixar.

No Linux, alternativamente, você pode fazer todo o processo de download e extração do modelo de língua portuguesa na pasta do projeto através dos comandos:

cd pasta_do_projeto
wget https://alphacephei.com/vosk/models/vosk-model-small-pt-0.3.zip
unzip vosk-model-small-pt-0.3.zip
mv vosk-model-small-pt-0.3 modelo

Convertendo áudio de arquivo para texto

Vamos começar pelo script mais simples, para converter áudio para texto diretamente de um arquivo wave. Vale destacar que a Vosk trabalha com arquivos wave de 16 bits de um canal. Arquivos gravados em outro formato não serão adequadamente processados.

O código fonte completo é assim:

import wave
import json
from vosk import Model, KaldiRecognizer, SetLogLevel 
SetLogLevel(-1)

modelo = Model("model")

def fala_para_texto(arquivo):
  
    with wave.open(arquivo, "rb") as arq:
        rec = KaldiRecognizer(modelo, arq.getframerate())            
        while True:
            data = arq.readframes(4000)
            rec.AcceptWaveform(data)
            if len(data) == 0:
                break
                        
    resultado = json.loads(rec.FinalResult())
    return resultado["text"]    

if __name__ == "__main__":
    print(fala_para_texto("test.wav"))    

A primeira linha depois das importações de pacotes, SetLogLevel(-1) serve para que o Vosk não imprima na tela do terminal as mensagens de log (experimente tirar essa linha para saber do que estou falando).

Na linha seguinte, inicializamos nosso modelo para reconhecimento de fala instanciando a classe Model (lembre de extrair o modelo que você baixou dentro da pasta “modelo”!).

Então, definimos nossa função fala_para_texto() que lerá um arquivo wave e retornará uma string com a transcrição da fala encontrada. A função wave.open() é muito semelhante ao open() do Python, podendo inclusive ser tratada como um gerenciador de contexto, isto é, podemos utilizar a palavra-chave with para estabelecer um contexto de execução.

Depois de carregar o arquivo, inicializamos o reconhecedor de fala, instanciando a classe KaldiRecognizer, que ligamos à variável rec. Os parâmetros de inicialização desse objeto são o modelo que acabamos de criar e a taxa de amostragem do áudio, que diz basicamente quantas amostragens foram coletadas por segundo na gravação do áudio. Obtemos essa informação chamando o método getframerate() do objeto que representa nosso arquivo wave.

Então fazemos o reconhecimento propriamente dito dentro de um loop infinito que lê uma determinada quantidade de quadros (amostras) do arquivo com o método readframes() e alimenta o reconhecedor de fala utilizando o método chamado AcceptWaveform(). Esse método recebe os dados e faz o processamento do áudio, retornando True, se ele tiver detectado uma fala inteira, e False se na etapa atual do processamento não for possível terminar a conversão de fala pra texto.

A qualquer momento, podemos checar o resultado parcial do processamento com a função PartialResult(), que retorna uma string contendo o texto convertido até o momento. E quando AcceptWaveform() retornar True, podemos chamar o método FinalResult(), que retorna a string com o resultado final do processamento. Em nosso caso específico, não precisamos testar o valor de AcceptWaveform() já que ao final do arquivo, ele terá chegado a um resultado de todo modo (se ele não detectar nenhuma fala, o resultado final será uma string vazia).

A string retornada por PartialResult() ou FinalResult() segue o formato json, isto é, é uma string que contém um conjunto de pares chave/valor em forma de texto, muito semelhantes aos dicionários do Python. O texto transcrito está contido na chave “text” da string json. Podemos então extrair o texto dessa string utilizando a função loads() do módulo json do Python, que transforma a string em um dicionário. Então nossa função retorna o valor da chave “text”.

Se você não entendeu pra que serve aquele if logo depois da função, dá uma olhada nesse post aqui!

Agora, basta mudar o nome do arquivo para testar o script com seus áudios. Simples não? Agora vamos para um programinha um pouco mais complexo!

Convertendo áudio para texto em tempo real

O lado bom de converter áudio para texto em tempo real é que não precisamos nos preocupar com armazenagem de áudio em disco, mas essa abordagem também tem uma desvantagem: a precisão da conversão não costuma ser tão boa quanto se estivéssemos trabalhando com um arquivo wave, mas o resultado costuma ser razoável. E se você tiver um bom equipamento, pode obter resultados melhores que eu (com o microfone do notebook!).

Captando o áudio do microfone com sounddevice

Comecemos entendendo como captamos o áudio do microfone com a biblioteca sounddevice. Essa biblioteca oferece algumas opções diferentes para captação de entrada e produção de saída de áudio.

O esqueleto de um programa para capturar áudio seria mais ou menos assim:

import queue
import sounddevice as sd

q = queue.Queue()

def callback(indata, frames, time, status):
    q.put(bytes(indata))

device_info = sd.query_devices(sd.default.device[0], 'input')
samplerate = device_info['default_samplerate']
stream = sd.RawInputStream(dtype='int16', channels=1, callback=callback)

stream.start()

try:
    while True:
        data = q.get()

except KeyboardInterrupt:
    stream.close()
    print("Encerrando a captura de áudio...")

Agora vamos ver passo a passo como nosso programa funciona.

Aqui, utilizamos a classe RawInputStream e seus métodos para captar o áudio do microfone e armazená-los em buffers internos do Python. Um objeto dessa classe funciona capturando dados do dispositivo de entrada e enviando para uma função de callback definida por nós, que irá consumir os dados capturados em uma thread separada.

Nota: Uma thread, grosso modo, nada mais é que um fluxo de execução. Um programa tradicional com uma única thread possui apenas um fluxo de execução, enquanto programas com múltiplas threads possuem dois ou mais fluxos rodando separadamente (é como se o seu programa fizesse mais de uma coisa ao mesmo tempo ao invés de seguir uma sequência única de instruções).

Vamos pular por um momento o início do código com a instanciação de uma queue e a definição da função de callback para entender melhor a inicialização da RawInputStream. Esse método aceita uma série de argumentos, muitos deles opcionais. No nosso caso, para instanciá-la, vamos fornecer alguns argumentos básicos. Esses detalhes técnicos podem não fazer muito sentido pra quem não entende muito de sinais digitais. Se esse for o seu caso (é o meu!), não se importe tanto. Você não vai precisar se aprofundar no assunto para seguir o post. Os argumentos que iremos utilizar são:

• samplerate (taxa de amostragem): Já vimos do que se trata no primeiro script. Dessa vez, descobrimos essa informação através da função query_devices() da sounddevice.
• dtype: o formato utilizado pelo buffer de amostragem (utilizado pela função de callback). Algumas opções são possíveis, mas o formato utilizado pelo Vosk é o ‘int16’, pois ele utiliza amostragens de 16 bits.
• channels: o número de canais de som. Utilizaremos apenas 1.
• callback: a função de callback definida por nós para consumir os dados coletados, que explicarei melhor em breve.

Para descobrir a taxa de amostragem, primeiro descobrimos qual é o dispositivo de entrada padrão, cujo identificador está contido na variável sd.default.device[0] (o dispositivo de saída padrão estaria em sd.default.device[1]).

Com essa informação, podemos chamar o método sd.query_devices(), que recebe como argumento o identificador do dispositivo e o tipo (input ou output) e retorna um dicionário com as informações do dispositivo em questão. A linha device_info = sd.query_devices(sd.default.device[0], ‘input’), coloca o dicionário com as informações do dispositivo de entrada padrão na variável device_info.

Finalmente, achamos a taxa de amostragem na entrada “default_samplerate” do dicionário.

Agora vamos entender a função de callback. Essa função é chamada no background por nosso objeto RawInputStream com quatro argumentos: indata, frames, time e status. Para nossa aplicação, só nos interessa o primeiro, que contém os dados coletados pela stream, mas recomendo a leitura da documentação do sounddevice (alerta: está em inglês!) para que você tenha uma noção sobre o significado dos outros argumentos.

Nossa função de callback simplesmente pegará os dados coletados, converterá para bytes (originalmente, trata-se de um formato específico de buffer) e enviará para um objeto do tipo queue que definimos bem no início do código.

A nossa Queue, contida no módulo queue, implementa uma fila, uma estrutura de dados em que o primeiro elemento inserido é o primeiro a ser removido. Mas o maior motivo para utilizarmos uma queue em nosso programa é que a queue implementada nesse módulo permite que programas com múltiplos threads (nosso caso!) compartilhem informações de forma consistente entre suas threads.

Os métodos mais relevantes da queue são o put(), que coloca um certo objeto na fila e o get(), que pega o objeto colocado na fila há mais tempo.

Agora que nossa stream foi instanciada e a função de callback foi definida do modo apropriado, podemos dar início à captura de áudio. Fazemos isso chamando o método start() da stream. A partir daí entramos em um laço infinito que consome os dados enviados para a fila na linha data = q.get() (embora a gente não faça nada com os dados).

Vale lembrar que ao mesmo tempo que esse laço é executado, existe uma outra thread capturando a entrada do microfone. Quando o usuário pressiona Ctrl-C (interrupção de teclado) o programa suspende a captura de áudio com o método close() da stream e encerra a execução.

É importante notar que as classes do tipo Stream do sounddevice também são gerenciadores de contexto, ou seja, também podemos utilizar a palavra-chave with para estabelecer um contexto de execução. Nesse caso, não precisamos nos preocupar em iniciar e encerrar a nossa stream diretamente. Então, nosso código poderia ficaria assim:

import queue
import sounddevice as sd

q = queue.Queue()

def callback(indata, frames, time, status):
    q.put(bytes(indata))

device_info = sd.query_devices(sd.default.device[0], 'input')
samplerate = device_info['default_samplerate']

with sd.RawInputStream(dtype='int16', channels=1, callback=callback):

    try:
        while True:
            data = q.get()

    except KeyboardInterrupt:
    	print(“Encerrando a captura de áudio...”)

Realizando a conversão da fala para texto em tempo real

Depois de superada essa parte, o resto é bastante simples. Primeiro, vamos importar os elementos necessários da biblioteca vosk para utilizar o reconhecedor e, como no exemplo anterior, inicializar nosso modelo para reconhecimento de fala instanciando a classe Model, e depois criar um objeto da classe KaldiRecognizer ligado à variável rec,passando como argumentos o modelo e a taxa de amostragem.

modelo = Model("modelo")  
rec = KaldiRecognizer(modelo, samplerate) 

Você pode colocar esse código imediatamente depois do início do bloco with no código anterior. Então fazemos o reconhecimento da fala com AcceptWaveform(). Mas dessa vez, vamos enviar como argumento os dados coletados da fila e contidos na variável data.

Lembre que esse método retorna True, se ele tiver detectado uma fala completa, e False caso contrário. Então podemos testar esse valor de retorno para saber se já devemos imprimir a fala na tela. E pronto! O código completo fica assim:

import queue			#fila de mensagens
import sounddevice as sd	#módulo para captura de som do mic

#reconhecimento de fala
from vosk import Model, KaldiRecognizer, SetLogLevel 

SetLogLevel(-1)			#desabilita os logs do vosk

q = queue.Queue()			#instanciação da fila

#Função usada pelo RawInputStream
def callback(indata, frames, time, status):
    q.put(bytes(indata))

#Descobrir a taxa de amostragem
device_info = sd.query_devices(sd.default.device[0], 'input')
samplerate = device_info['default_samplerate']

with sd.RawInputStream(dtype='int16', channels=1, callback=callback):
    #Inicializa o modelo e o reconhecedor
    modelo = Model("modelo")
    rec = KaldiRecognizer(modelo, samplerate)
    try:
        while True:
            #Coleta os dados e testa se o texto foi
            #convertido com sucesso
            data = q.get()
            if rec.AcceptWaveform(data):
                print(rec.FinalResult())
    except KeyboardInterrupt:
        print("Encerrando a captura de áudio…")

Pronto! Ao rodar o programa, se você tiver feito tudo direitinho, você vai ver algumas mensagens de log do vosk, e, em seguida, começará a aparecer na tela as palavras detectadas pelo programa.

Se quiser imprimir apenas o texto, ao invés da string json, siga os passos do primeiro código para converter o json para um dicionário (é bom pra exercitar!).

Para finalizar o post, como prometido, segue um pequeno script que capta o áudio do microfone em tempo real, converte para texto e depois converte novamente para voz:

import queue
import sounddevice as sd
import vosk
import json
import pyttsx3

#Fila de mensagens
q = queue.Queue()

#Função chamada pelo RawStreamInput
def callback(indata, frames, time, status):
    q.put(bytes(indata))

#Acha taxa de amostragem do dispositivo padrão
device_info = sd.query_devices(sd.default.device[0], 'input')
samplerate = device_info['default_samplerate']

#Configuração da engine de síntese de voz
engine = pyttsx3.init()
voices = engine.getProperty("voices")
engine.setProperty("voice", "brazil") 
engine.setProperty("rate", 120)

#Inicializa a stream em um novo contexto
with sd.RawInputStream(dtype='int16', channels=1, callback=callback) as stream:
    modelo = vosk.Model("modelo")
    rec = vosk.KaldiRecognizer(modelo, samplerate)
    try:
        while True:
            data = q.get()
            if rec.AcceptWaveform(data):
                #Converte a string de FinalResult() para json
                result = json.loads(rec.FinalResult())
               	text = result["text"]    
                #Se existe um texto para ser falado
                #Pare a captação de áudio (para o programa não detectar
                #a fala que ele mesmo produz), e depois de sintetizar a fala
                #continue a captura de áudio.
               	if text:
                    stream.stop()
                    engine.say(text)
                    engine.runAndWait()
                    stream.start()
    except KeyboardInterrupt:
        print("Encerrando a captura de áudio...")

Aqui temos um vídeo de exemplo do script:

Melhorando a precisão de sua aplicação

Como eu falei no início do post, a precisão da conversão nem sempre é muito boa. Uma forma de contornar isso é modificando o modelo de linguagem utilizado, de modo a adaptá-lo para o tipo de aplicação que você deseja, mas essa é uma tarefa relativamente complexa, que exige certos conhecimentos técnicos que eu, particularmente, não tenho!

Se você estiver criando uma aplicação simples, que deve identificar um número reduzido de palavras, como por exemplo um script para um robô quadrúpede de estimação (por que não?), você pode alterar o vocabulário do reconhecedor com as palavras que você deseja reconhecer no momento de sua inicialização. Para isso, basta passar como terceiro argumento uma lista de strings em formato json com essas palavras. Por exemplo:

rec = KaldiRecognizer(model, wf.getframerate(), '["senta", "levanta", "rola", "finge", "de", "morto"]')

Você vai reparar que apenas as palavras pertencentes a essa lista são reconhecidas. Mas a precisão para reconhecê-las aumenta!

Espero que você tenha gostado do post, continue acompanhando o blog para aprender mais sobre Python e programação em geral!