Всем привет! После прочтения постов про голосового ассистента (первый, второй) и сервис 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.