Telegram бот с offline распознаванием голосовых и генерацией аудио из текста
Telegram бот с offline распознаванием голосовых и генерацией аудио из текста

Всем привет! После прочтения постов про голосового ассистента (первый, второй) и сервис Silero, мне стало интересно поиграться с offline распознаванием аудио, а также обратным преобразованием текста в аудио. И как все начинающие разработчики я сделал своего Telegram бота. Просто Telegram – это удобный и мобильный интерфейс для взаимодействия с чем угодно. В своем пет-проекте я применил: Python, aiogram, Vosk, Silero и ffmpeg.

Этот проект не призван заменить существующие решения. Просто им было интересно заниматься в свободное время и учиться писать код на Python. Проект работает под Windows, но с небольшими изменениями запустится и под другими ОС.

Главное отличие проекта от примеров из Vosk и Silero – легкость использования, т.к. вся инициализация параметрами по умолчанию проходит в отдельном классе. Остается лишь создать экземпляр класса и передать в него текст или путь до аудио файла.

Под offline работой подразумевается, что проект полностью работает на одной машине и не использует сторонние сервисы, кроме Telegram. Можно не запускать Telegram бота, а только скачать модели и отдельно поиграться с распознаванием или генерацией аудио.

Для удобства работы я разбил проект на 3 файла:

  1. bot.py – код Telegram бота

  2. stt.py – код преобразования аудио в текст, Speech to Text

  3. tts.py – код преобразования текста в аудио, Text to Speech

Код Telegram бота

Бот построен на асинхронной библиотеке для Telegram – aiogram, которая позволяет достаточно легко создать скрипт простого бота для чтения и отправки сообщений.

Бот в Telegram – это как отдельный пользователь, которым можно управлять через API. Его нужно создать в самом Telegram и получить токен. Процесс создания и получения токена не описываю, т.к. это легко найти в интернете.

Ниже представлен пример скрипта бота.

TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN")

bot = Bot(token=TELEGRAM_TOKEN)  # Объект бота с токеном
dp = Dispatcher(bot)  # Диспетчер для бота

# Хэндлер на команду /start , /help
@dp.message_handler(commands=["start", "help"])
async def cmd_start(message: types.Message):
    await message.reply(
        "Привет! Это Бот для конвертации голосового/аудио сообщения в текст"
        " и создания аудио из текста."
    )

# Запуск бота
if __name__ == "__main__":
    try:
        executor.start_polling(dp, skip_updates=True)
    except (KeyboardInterrupt, SystemExit):
        pass

Хэндлер – это функция, реагирующая на какое-то событие. В нашем случае на команду /start или /help бот ответит сообщением.

По сути наш бот должен реагировать на получение сообщения с текстом и аудио. Поэтому добавим хэндлеры для реагирования на получение текста и аудио.

tts = TTS()

# Хэндлер на получение текста
@dp.message_handler(content_types=[types.ContentType.TEXT])
async def cmd_text(message: types.Message):
    await message.reply("Текст получен")

    out_filename = tts.text_to_ogg(message.text)

    # Отправка голосового сообщения
    path = Path("", out_filename)
    voice = InputFile(path)
    await bot.send_voice(message.from_user.id, voice,
                         caption="Ответ от бота")

    os.remove(out_filename) # Удаление временного файла

При получении текста мы отправляем ответ, что получили текст. Затем конвертируем текст в аудио и отправляем голосовое сообщение. В конце удаляем временный файл.

Хэндлер для аудио сообщения.
stt = STT()

# Хэндлер на получение голосового и аудио сообщения
@dp.message_handler(content_types=[
    types.ContentType.VOICE,
    types.ContentType.AUDIO,
    types.ContentType.DOCUMENT
    ]
)
async def voice_message_handler(message: types.Message):
    """
    Обработчик на получение голосового и аудио сообщения.
    """
    if message.content_type == types.ContentType.VOICE:
        file_id = message.voice.file_id
    elif message.content_type == types.ContentType.AUDIO:
        file_id = message.audio.file_id
    elif message.content_type == types.ContentType.DOCUMENT:
        file_id = message.document.file_id
    else:
        await message.reply("Формат документа не поддерживается")
        return

    file = await bot.get_file(file_id)
    file_path = file.file_path
    file_on_disk = Path("", f"{file_id}.tmp")
    await bot.download_file(file_path, destination=file_on_disk)
    await message.reply("Аудио получено")

    text = stt.audio_to_text(file_on_disk)
    if not text:
        text = "Формат документа не поддерживается"
    await message.answer(text)

    os.remove(file_on_disk)  # Удаление временного файла

Этот хэндлер реагирует на голосовые, аудио и любые документы, т.к. не понятно какой файл пришлет пользователь. Скрипт скачивает файл на диск. В описании aiogram написано, что файл должен быть менее 20 Мб иначе придется скачать другим способом. Я пробовал отправлять короткие сообщения, так что не проверял это. Файл с аудио конвертируем в текст и отправляем ответ. В конце удаляем временный файл.

Код распознавания аудио, Speech to Text

Распознавать аудио будем с помощью offline проекта Vosk. Нам нужно только скачать модель для нейросети и поместить в папку models/vosk. Модель представляет собой архив. Просто создайте папку model и поместите содержимое архива туда, а затем закиньте папку model в models/vosk.

В файле stt.py есть класс, который проводит инициализацию модели Vosk параметрами по умолчанию и предоставляет метод для конвертации аудио в текст. Vosk может работать только с wav файлами, поэтому тут используют ffmpeg для конвертации.

Ffmpeg - набор open-source библиотек для конвертирования аудио- и видео в различных форматах. Для Windows скачайте набор exe файлов с сайта проекта и поместите файл ffmpeg.exe в папки models/vosk и models/silero.

class STT:
    """
    Класс для распознавания аудио через Vosk и преобразования его в текст.
    Поддерживаются форматы аудио: wav, ogg
    """
    default_init = {
        "model_path": "models/vosk/model",  # путь к папке с файлами STT модели Vosk
        "sample_rate": 16000,
        "ffmpeg_path": "models/vosk"  # путь к ffmpeg
    }

    def __init__(self,
                 model_path=None,
                 sample_rate=None,
                 ffmpeg_path=None
                 ) -> None:
        """
        Настройка модели Vosk для распознавания аудио и
        преобразования его в текст.

        :arg model_path:  str  путь до модели Vosk
        :arg sample_rate: int  частота выборки, обычно 16000
        :arg ffmpeg_path: str  путь к ffmpeg
        """
        self.model_path = model_path if model_path else STT.default_init["model_path"]
        self.sample_rate = sample_rate if sample_rate else STT.default_init["sample_rate"]
        self.ffmpeg_path = ffmpeg_path if ffmpeg_path else STT.default_init["ffmpeg_path"]

        self._check_model()

        model = Model(self.model_path)
        self.recognizer = KaldiRecognizer(model, self.sample_rate)
        self.recognizer.SetWords(True)

    
    def audio_to_text(self, audio_file_name=None) -> str:
        """
        Offline-распознавание аудио в текст через Vosk
        :param audio_file_name: str путь и имя аудио файла
        :return: str распознанный текст
        """
        # Конвертация аудио в wav и результат в process.stdout
        process = subprocess.Popen(
            [self.ffmpeg_path,
             "-loglevel", "quiet",
             "-i", audio_file_name,          # имя входного файла
             "-ar", str(self.sample_rate),   # частота выборки
             "-ac", "1",                     # кол-во каналов
             "-f", "s16le",                  # кодек для перекодирования, у нас wav
             "-"                             # имя выходного файла нет, тк читаем из stdout
             ],
            stdout=subprocess.PIPE
                                   )

        # Чтение данных кусками и распознование через модель
        while True:
            data = process.stdout.read(4000)
            if len(data) == 0:
                break
            if self.recognizer.AcceptWaveform(data):
                pass

        # Возвращаем распознанный текст в виде str
        result_json = self.recognizer.FinalResult()  # это json в виде str
        result_dict = json.loads(result_json)    # это dict
        return result_dict["text"]               # текст в виде str

Метод STT.audio_to_text принимает путь до аудио файла. Ffmpeg конвертирует файл в нужный формат, а затем модель Vosk через recognizer распознает текст. В конце возвращаем распознанный текст.

Кстати, если знаете рабочий способ, как в ffmpeg передать байты и получить результат в байтах без сохранения файла на диск, то напишите в комментариях.

Вариант применения:

stt = STT()
print(stt.audio_to_text("test-1.ogg"))

Код генерации аудио, Text to Speech

Генерировать аудио будем с помощью offline модели проекта Silero. Нам нужна модель для нейросети, но при первом запуске она сама скачается по URL из настроек по умолчанию (замените URL, если вам нужна другая модель). Также нам нужно скачать ffmpeg и положить в папку models/silero.

В файле tts.py есть класс, который проводит инициализацию модели Silero параметрами по умолчанию и предоставляет метод для конвертации текста в аудио. По умолчанию Silero генерирует аудио в wav, но т.к. он занимает много места, то я конвертировал wav в ogg через ffmpeg.

class TTS:
    """
    Класс для преобразования текста в аудио.
    Поддерживаются форматы аудио: wav, ogg
    """
    default_init = {
        "sample_rate": 24000,
        "device_init": "cpu",
        "threads": 4,
        "speaker_voice": "kseniya",
        "model_path": "models/silero/model.pt",  # путь к файлу TTS модели Silero
        "model_url": "https://models.silero.ai/models/tts/ru/v3_1_ru.pt", # URL к TTS модели Silero
        "ffmpeg_path": "models/silero"  # путь к ffmpeg
    }

    def __init__(
        self,
        sample_rate=None,
        device_init=None,
        threads=None,
        speaker_voice=None,
        model_path=None,
        model_url=None,
        ffmpeg_path=None
    ) -> None:
        """
        Настройка модели Silero для преобразования текста в аудио.

        :arg sample_rate: int       # 8000, 24000, 48000 - качество звука
        :arg device_init: str       # "cpu", "gpu"(для gpu нужно ставить другой torch)
        :arg threads: int           # количество тредов, например, 4
        :arg speaker_voice: str     # диктор "aidar", "baya", "kseniya", "xenia", "random"(генерит голос каждый раз, долго)
        :arg model_path: str        # путь до модели silero
        :arg model_url: str         # URL к TTS модели Silero
        :arg ffmpeg_path: str       # путь к ffmpeg
        """
        self.sample_rate = sample_rate if sample_rate else TTS.default_init["sample_rate"]
        self.device_init = device_init if device_init else TTS.default_init["device_init"]
        self.threads = threads if threads else TTS.default_init["threads"]
        self.speaker_voice = speaker_voice if speaker_voice else TTS.default_init["speaker_voice"]
        self.model_path = model_path if model_path else TTS.default_init["model_path"]
        self.model_url = model_url if model_url else TTS.default_init["model_url"]
        self.ffmpeg_path = ffmpeg_path if ffmpeg_path else TTS.default_init["ffmpeg_path"]

        self._check_model()

        device = torch.device(self.device_init)
        torch.set_num_threads(self.threads)
        self.model = torch.package.PackageImporter(self.model_path).load_pickle("tts_models", "model")
        self.model.to(device)

   def _get_wav(self, text: str, speaker_voice=None, sample_rate=None) -> str:
        """
        Конвертирует текст в wav файл

        :arg text:  str  # текст до 1000 символов
        :arg speaker_voice:  str  # голос диктора
        :arg sample_rate: str  # качество выходного аудио
        :return: str  # путь до выходного файла
        """
        if text is None:
            raise Exception("Передайте текст")

        # Удаляем существующий файл чтобы все хорошо работало
        if os.path.exists("test.wav"):
            os.remove("test.wav")

        if speaker_voice is None:
            speaker_voice = self.speaker_voice

        if sample_rate is None:
            sample_rate = self.sample_rate

        # Сохранение результата в файл test.wav
        return self.model.save_wav(
            text=text,
            speaker=speaker_voice,
            sample_rate=sample_rate
        )

    def text_to_ogg(self, text: str, out_filename: str = None) -> str:
        """
        Конвертирует текст в файл ogg.
        Модель игнорирует латиницу, но поддерживает цифры числами.

        :arg text: str  # текст кирилицей
        :return: str    # имя выходного файла
        """
        if text is None:
            raise Exception("Передайте текст")

        # Делаем числа буквами
        text = self._nums_to_text(text)

        # Генерируем ogg если текст < 1000 символов
        if len(text) < 1000:
            # Возвращаем путь до ogg
            ogg_audio_path = self._get_ogg(text)

            if out_filename is None:
                return ogg_audio_path

            return self._rename_file(ogg_audio_path, out_filename)

        # Разбиваем текст, конвертируем и склеиваем аудио в один файл
        texts = [text[x:x+990] for x in range(0, len(text), 990)]
        files = []
        for index in range(len(texts)):
            # Конвертируем текст в ogg, возвращаем путь до ogg
            ogg_audio_path = self._get_ogg(texts[index])
            # Переименовываем чтобы не затереть файл
            new_ogg_audio_path = f"{index}_{ogg_audio_path}"
            os.rename(ogg_audio_path, new_ogg_audio_path)
            # Добавляем новый файл в список
            files.append(new_ogg_audio_path)

        # Склеиваем все ogg файлы в один
        ogg_audio_path = self._merge_audio_n_to_1(files, out_filename="test_n_1.ogg")
        # Удаляем временные файлы
        [os.remove(file) for file in files]

        if out_filename is None:
            return ogg_audio_path

        return self._rename_file(ogg_audio_path, out_filename)

Класс получился большим, т.к. пришлось бороться с ограничениями:

  • До 1000 символов на входе модели

  • Сохранение аудио только в виде файла с названием text.wav

  • Не восприимчивость цифр, пришлось преобразовать их в текст

  • Модель русского языка не говорит на английском

Модель русского языка не может озвучить английский текст, поэтому такой текст просто игнорируется. Пробовал делать транслитерацию (английские символы русскими буквами), но это звучит ужасно. Тут нужно делать перевод с английского на русский, либо озвучивать отдельно и потом склеивать результат или просто использовать модель, которая поддерживает 2 языка. В общем не присылайте слова на английском, они будут проигнорированы.

Основная магия происходит во внутреннем методе TTS._get_wav(), который принимает текст и необязательные параметры: голос спикера и качество выходного аудио. Модель создает аудио файл с именем text.wav и метод возвращает путь до него.

Для работы с классом TTS есть основные методы TTS.text_to_ogg() и TTS.text_to_wav(). Методы принимают текст любой длины и необязательный параметр имя выходного файла. Как раз в этих методах происходит преобразование цифр в текст, а также разбивка текста на строки до 990 символов. Сделал самое простое разбиение, но по-хорошему нужно делить по предложениям или словам, т.к. это влияет на конечный результат.

Отдельные строки по 990 символов озвучиваю моделью, а потом склеиваю через ffmpeg, что на выходе метода всегда возвращает 1 файл.

Вариант применения:

tts = TTS()
print(tts.text_to_ogg("Привет,Хабр! Тэст 1 2 три четыре", "test-1.ogg"))

TODO

В проекте еще многого чего можно улучшить, вот несколько моментов:

  • Реализовать методы без сохранения файлов на диск

  • Добавить перевод английских слов на русские

  • Упаковать проект в докер-контейнер для развертывания на сервере

  • Проверить ограничения по размеру сообщения и файла в Telegram

  • Когда-нибудь написать тесты

Выводы

Возможно у меня не получилось с первого раза промышленное решение для крупных корпораций, но думаю такой пет-проект позволил мне попрактиковаться сразу в нескольких областях программирования:

  • Продумать архитектуру проекта

  • Написать своего Telegram бота

  • Получить опыт на реальном проекте, который потом еще можно развивать

  • Развить умение гуглить ошибки и читать документацию

В общем занимайтесь интересными вещами и получайте от этого удовольствие, даже если это нужно только вам.

Ссылка на проект в git.

Комментарии (14)


  1. port20031
    21.10.2022 04:10

    Доброго времени суток .

    Если не тяжело , выложи пожалуйста, результат вывода русского текста в вав .

    Играюсь в систему умный дом и голос формирую RHVoice , хочу сравнить качество


    1. foxairman Автор
      21.10.2022 09:07

      Здравствуйте! Вот ссылка на 3 файла wav с разным качеством. Silero поддерживает битрейт 8000, 24000 и 48000.


      1. port20031
        21.10.2022 20:17

        Еще вопрос и даже очень шкурный ))))


        Silero нужен интернет ?


        1. foxairman Автор
          21.10.2022 20:23

          Для Silero и Vosk не нужен. Интернет нужен только для Telegram


          1. port20031
            22.10.2022 23:00

            Столкнулся с проблемой озвучки чисел и знаков +,-
            Мой локальный говорит погоду и время .
            Как решить эту задачу ?


            1. foxairman Автор
              23.10.2022 00:36

              Я просто менял их на текст, например, этот код делает числа словами

              # Делаем числа буквами
              text = self._nums_to_text(text)


  1. Zhuikoff
    21.10.2022 06:08

    Работа с ffmpeg dll напрямую есть, но на паскале.


  1. S-trace
    21.10.2022 08:54
    +2

    Я тоже игрался с Silero TTS и выяснил, что разные голоса имеют разное ограничение на максимальную длину обрабатываемого текста, 1000 символов уверенно тянет только eugene, а kseniya надо передавать не больше895 символов (при этом некоторые тексты большего размера она может обработать, а другие - нет). Проверено тестами.

    Взгляните на скрипт https://github.com/S-trace/silero_tts_standalone/blob/master/tts.py - возможно найдёте что-нибудь полезное для себя в def preprocess_text - эта функция обрабатывает текст и затем разделяет текст по границам предложений на чанки пригодные для озвучивания.


  1. Jury_78
    21.10.2022 10:06
    +1

    как в ffmpeg передать байты и получить результат в байтах

    Может я не правильно понял... Почему бы не использовать поток (stream)?

    Пакет ffmpeg-python (пример для видео):

    import ffmpeg
    stream = ffmpeg.input('input.mp4')
    stream = ffmpeg.hflip(stream)
    stream = ffmpeg.output(stream, 'output.mp4')
    ffmpeg.run(stream)


    1. foxairman Автор
      21.10.2022 18:41

      Я хотел избавится от сохранения файлов на диск. В примере 'input.mp4' скорее всего лежит на диске, а я хотел чтобы он был в оперативной памяти, что-то типа временного файла в виде io.BytesIO или что-то такое. Это позволит избавиться от постоянного сохранения и удаления файлов на диске при конвертации через ffmpeg. И думаю вся система будет работать быстрее


      1. Jury_78
        23.10.2022 20:29
        +1

        Можно попробовать использовать mmap.


        1. foxairman Автор
          23.10.2022 23:42

          Вроде, нашел способ передавать в байтах и получать в байтах без сохранения файла на диск, код будет типа такого. Попробую еще с помощью mmap, спасибо!

          def wav_to_ogg_bytes(in_bytes: bytes) -> bytes:
              command = [
                  "ffmpeg",
                  "-i", 'pipe:0',         # stdin
                  "-f", "ogg",            # format
                  "-acodec", "libvorbis", # codec
                  "pipe:1"                # stdout
              ]
              proc = subprocess.Popen(
                  command,
                  stdin=subprocess.PIPE,
                  stdout=subprocess.PIPE,
                  stderr=subprocess.PIPE
              )
              out_bytes, err = proc.communicate(input=in_bytes)
              return out_bytes


  1. Rasters
    21.10.2022 18:29
    +1

    Кстати, если знаете рабочий способ, как в ffmpeg передать байты и получить результат в байтах без сохранения файла на диск, то напишите в комментариях.

    Посмотрите для примера
    потребуется: https://github.com/kkroening/ffmpeg-python


  1. S0mbre
    23.10.2022 02:33
    +1

    Спасибо за статью! Я бы только в качестве совета порекомендовал реализовать основные методы tts, stt асинхронно, т.к. основная работа бота выполняется именно там, в этом а иначе мало смысла в использовании async пакетов.