Привет, Хабр!
Время от времени я возвращаюсь к своему pet-проекту голосового ассистента с кодовым именем «Альфа», который разрабатывался как приватный голосовой интерфейс (а-ля «умная колонка») для управления своим «Умным домом». И в этот раз – так сошлись звезды или под влиянием магнитных бурь – мне очень захотелось добавить новый навык. А что из этого вышло, читайте далее.
Друзья, чтобы понимать, что тут вообще происходит, рекомендую ознакомиться с моими ранними статьями: «Моя б̶е̶з̶умная колонка или бюджетный DIY голосового ассистента для умного дома» – тут и «Моя б̶е̶з̶умная колонка: часть вторая // программная» – тут. В статьях описана аппаратная и базовая программная реализация «Альфы». Спасибо!
❯ М̶о̶и̶ ̶х̶о̶т̶е̶л̶к̶и̶ Техническое задание
Прежде чем продолжить, давайте вспомним что такое навык. Навык (Skill) – это сторонняя программа, которая подключается к голосовому ассистенту (через специальный API или модуль) и расширяет его функционал. Она активируется определенной фразой-триггером.
В моем случае мне необходимо реализовать функционал навыка, который бы обеспечивал запись планируемых событий с помощью голоса в какой-нибудь локальный или удаленный сервис планировщика для возможности синхронизации задач с устройствами пользователя (например, для просмотра событий на смартфоне пользователя). Касательно последнего, то проще всего для этих целей интегрировать Google календарь или аналогичные сервисы. И для реализации данного навыка нам потребуется сделать следующее:
Добавить в словарь команды: для активации навыка и вывода планируемых событий;
Разработать логику активации навыка и ввода голосовых данных;
Разработать метод извлечения параметров (название события, дату и время) из голосовых данных (после транскрибации);
Разработать модуль взаимодействия с внешним сервисом (Google календарь);
Разработать метод вывода сохраненных задач с помощью голосового оповещения (синтеза речи).
Для лучшего понимания моей фантазии, ниже представлена блок-схема логики навыка.
❯ Команды активации и действия
В прошлой статье я уже рассказывал о словаре команд, где указаны ключи активации и варианты произношения. Теперь нам необходимо его немного скорректировать, добавив пару команд, изменив его до следующего вида:
command_dic = { #... предыдущие команды # Новые команды "schedule_list": ('скажи задачи на сегодня', 'какой список задач сегодня', 'список дел на сегодня', 'список дел', 'какие задачи сегодня'), "schedule_add": ('добавь задачу', 'добавь напоминание','создай событие', 'добавь событие') }
Как вы уже наверное смогли догадаться, ключ команды schedule_list — отвечает за вывод списка задач, а schedule_add — за добавление задачи. Теперь осталось только добавить в обработчик команд данные ключи:
def command_processing(key: str): match key: # ... Предыдущие команды case 'schedule_list': # Активация команды schedule_list shedule_list() case 'schedule_add': # Активация команды schedule_add # Здесь будет какой-то код case _: print('Нет данных')
Изначально у нас реализована следующая функция для распознания имени и обработки команд:
def response(voice: str): if glob_var.read_bool_wake_up(): # этап второй, распознавание команды command_processing(recognize_command(voice)) # распознавание и выполнение команды glob_var.set_bool_wake_up(False) # после выполнения команды, перехохим в режим распознования имени glob_var.set_bool_wake_up(name_recognize(voice)) # проверяем наличие имени в потоке if glob_var.read_bool_wake_up(): # если имя обнаружено, воспроизводим звуковой сигнал tts.play_wakeup_sound('notification.wav')
И так как нам нужна повторная активация ассистента (исключая функцию распознания имени) при активации навыка, то внесем небольшие изменения в выше указанный код:
stat = False # Глобальная переменнаая, статус активации навыка def response(voice: str): global stat if glob_var.read_bool_wake_up(): # этап второй, распознавание команды stat = command_processing(recognize_command(voice), voice) # распознавание и выполнение команды с получением булевого значения от функции glob_var.set_bool_wake_up(stat) # после выполнения команды, перехохим в режим распознования имени if not stat: # если навык активирован, то не проверяем наличие имени в потоке glob_var.set_bool_wake_up(name_recognize(voice)) # проверяем наличие имени в потоке if glob_var.read_bool_wake_up(): # если имя обнаружено, воспроизводим звуковой сигнал tts.play_wakeup_sound('notification.wav')
И, соответственно, чтобы функция обработка команд (command_processing) смогла нам возвращать булево значение, изменим её код:
def command_processing(key: str, voice: str): global stat result = False match key: # ... Предыдущие команды case 'schedule_list': # Активация команды schedule_list result = schedule_list() case 'schedule_add': # Активация команды schedule_add tts.speak("Хорошо, назовите имя события и время для добавления.") result = True case _: if stat: result = schedule_add(voice) # Пытаемся извлечь данные из строки else: print('Нет данных') return result # Возврат статуса функций планировщика
Также функция теперь принимает дополнительный параметр voice для последующей обработки и извлечения данных для записи в календарь.
❯ Интеграция Google календаря
Интеграция сервиса «Google календарь» достаточно простая и не представляет каких либо сложностей, у Гугла хорошие мануалы. Описывать полный процесс я не буду, так как это минимум на еще одну статью.
Чтобы интегрировать сервис в наш проект, нужно установить необходимые пакеты. Давайте же их скорее установим, команду для установки пакетов вы можете обнаружить ниже:
pip install google-api-python-client google-auth-httplib2 google-auth-oauthlib
Для взаимодействия с Google календарем, нам необходимо получить креды (credentials). Для сервиcного аккаунта (получается при создании приложения в панели Google Cloud) и для пользователя. Если первое — это проблема разработчика, то второе — на совести пользователя.
Так как «Альфа» по большей части имеет модульную структуру, то и взаимодействие с Google календарем будет реализовано в отдельном модуле. Ниже приведен код для работы с Google календарем (calendar_schedule.py):
calendar_schedule.py
import datetime import os.path from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow from googleapiclient.discovery import build from googleapiclient.errors import HttpError SCOPES = ['https://www.googleapis.com/auth/calendar'] calendarId = "primary" # Имя календаря пользователя class GoogleCalendar(): def __init__(self): creds = None if os.path.exists("token.json"): creds = Credentials.from_authorized_user_file("token.json", SCOPES) try: self.service = build("calendar", "v3", credentials=creds) except HttpError as error: print(f"An error occurred: {error}") # создание события в календаре def create_event(self, summary, start_time, end_time, description=None): """Создает новое событие в Google Календаре.""" try: # создание словаря с информацией о событии event_body = { 'summary': summary, 'start': {'dateTime': start_time.format(), 'timeZone': 'Asia/Yekaterinburg'}, # Укажите ваш часовой пояс 'end': {'dateTime': end_time.format(), 'timeZone': 'Asia/Yekaterinburg'}, 'description': description, } event = self.service.events().insert(calendarId=calendarId, body=event_body).execute() print(f'Событие создано: {event.get("htmlLink")}') return f'Событие {summary} добавлено в ваш календарь' except HttpError as error: print(f'Произошла нелепая ошибка: {error}') return f'Сожалею, но произошла ошибка при создании события. Попробуйте позже.' # вывод списка предстоящих событий def get_events_list(self): now_dts = datetime.datetime.now(tz=datetime.timezone.utc) now = datetime.datetime.now(tz=datetime.timezone.utc).isoformat() # Начальная дата(сейчас) ends = (now_dts + datetime.timedelta(days=1)).isoformat() # Конечная дата +1 день print('Получение списка следующих событий') events_result = self.service.events().list(calendarId=calendarId, timeMin=now, timeMax=ends, maxResults=10, singleEvents=True, orderBy='startTime').execute() events = events_result.get('items', []) return events
В модуле реализован отдельный класс GoogleCalendar() и методы create_event(), get_events_list() которые мы будем использовать для создания и получения событий.
Для работы модуля необходимо получить токен авторизации, который содержится в файле token.json. Для получения файла можно воспользоваться следующим скриптом (json_user.py):
json_user.py
import os.path from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow from googleapiclient.discovery import build from googleapiclient.errors import HttpError SCOPES = ['https://www.googleapis.com/auth/calendar'] calendarId = "primary" # Имя календаря пользователя def get_user_cred(): creds = None if os.path.exists("token.json"): creds = Credentials.from_authorized_user_file("token.json", SCOPES) # If there are no (valid) credentials available, let the user log in. if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: creds.refresh(Request()) else: flow = InstalledAppFlow.from_client_secrets_file( "credentials.json", SCOPES ) creds = flow.run_local_server(port=0) # Save the credentials for the next run with open("token.json", "w") as token: token.write(creds.to_json()) try: build("calendar", "v3", credentials=creds) except HttpError as error: print(f"An error occurred: {error}") if __name__ == "__main__": get_user_cred()
Файл credentials.json — это токен сервисного аккаунта, как я говорил ранее, он загружается из панели разработчика Google Cloud. Само собой, хранить токены в файлах — это не лучшая практика, но на данный момент и так сойдёт.
Запуск скрипта для получения файла выполняется с помощью команды:
python3 json_user.py
После запуска скрипта автоматически откроется браузер, где будет предложено выполнить вход с помощью аккаунта Google:
И заодно можно посмотреть предоставляемые разрешения:
❯ Извлекаем данные из фразы
Вот мы и добрались до самого интересного, а именно до извлечения данных для записи события из произнесенной фразы. На первый взгляд всё выглядит просто, но на самом деле — нет. Конечно, мы живем во времена искусственного интеллекта, «запусти LLM, «скорми» фразу и получи ответ в нужном формате» — скажите вы, да но тут несколько нюансов:
«Альфа» должна обрабатывать все запросы локально, чтобы обеспечивать приватность, быстродействие и независимость от внешних сервисов;
Локальное использование LLM требует больших вычислительных ресурсов, в том числе с применением NPU. «Альфа» работает на бюджетном железе, что ограничивает локальный запуск LLM;
Локальное применение LLM не обеспечит необходимого быстродействия.
Возможно я ошибаюсь, поправьте меня в комментариях, также буду рад вашим советам.
Учитывая всё вышесказанное, и ради быстродействия, будем использовать классический метод — парсинг с помощью регулярных выражений и стандартного Python-модуля re.
Работа с регулярными выражениями почему-то вызывает у меня дикую боль, поэтому делегируем эту боль задачу DeepSeek'у. И спустя несколько часов общения, мы получили более-менее рабочий код нашего парсера:
reminder_parser.py
Парсер возвращает необходимые нам данные в формате JSON. Для теста можно использовать следующий код:
Тест парсера
# Тестирование парсера def test_fixed_parser(): parser = SmartReminderParser() test_cases = [ # Фразы для теста "встреча в десять сорок пять", "позвонить маме в восемь ноль пять", "Позвонить в двадцать пять минут девятого", "Встреча в сорок пять минут второго", "Встреча в среду в десять сорок пять", "Сходить к врачу завтра в четырнадцать сорок пять", "Подготовить документы сегодня в двадцать три часа сорок пять минут", "оплатить счета сегодня в шестнадцать ноль ноль", "записаться к врачу сегодня в десять сорок пять", "принять лекарство в восемь утра и восемь вечера", "сходить в магазин в пятнадцать тридцать", "подготовить отчет в девятнадцать двадцать", "Сдать отчет в понедельник в пятнадцать тридцать", "Купить продукты сегодня в восемнадцать тридцать", ] print("Тестирование исправленного парсера с составными числами:") print("=" * 70) successful = 0 for phrase in test_cases: result = parser.parse(phrase) if result: print(f"✓ '{phrase}'") print(f" Действие: '{result['text']}'") print(f" Время: {result['time'].strftime('%d.%m.%Y %H:%M')}") print(f" Тип: {result['time_type']}") print() successful += 1 else: print(f"✗ '{phrase}' -> не распознано") # Диагностика text = parser._preprocess_text(phrase.lower()) print(f" После предобработки: '{text}'") print() print(f"Успешно распознано: {successful}/{len(test_cases)}") # Тест составных чисел print("\n" + "=" * 70) print("Тест составных числительных:") print("=" * 70) composite_tests = [ "двадцать пять", "сорок пять", "пятьдесят пять", "двадцать один", "тридцать восемь", "сорок два" ] for test in composite_tests: result = parser._word_to_num(test.replace(' ', '_')) print(f"'{test}' -> {result}") if __name__ == "__main__": test_fixed_parser()
Ниже видео живого теста парсера:
Youtube
Rutube
На видео показан промежуточный тест парсера, где в терминале отображается результат парсинга, а озвучивание произнесенной фразы выполняется для удобства, чтобы убедиться в корректности транскрибации в период отладки.
❯ Финальная интеграция
Если вы дочитали до этого момента, то поздравляю, мы близки к финалу :). Итак, давайте для понимания резюмируем то, что мы сделали выше. Мы написали два программных модуля, которые отвечают за работу с сервисом Google календарь (методы create_event(), get_events_list()) и за извлечение данных (парсинг) из произнесенной фразы (метод smart_parcer()). Теперь дело за малым – интегрировать наши программные модули в основной скрипт умной колонки.
Импортируем наши модули в основной скрипт:
import calendar_schedule import reminder_parser
И для проверки наличия авторизации в Google календаре, добавим следующий код:
errors_alarm = "" # Переменная для хранения ошибок для последкющего озвучивания try: schedule = calendar_schedule.GoogleCalendar() # Пытаемся инициировать класс работы с календарём except: print(f"Ошибка подключения календаря Google.") errors_alarm = "Ошибка подключения календаря."
И озвучиваем ошибки при запуске системы:
if errors_alarm: tts.speak('При запуске системы возникли следующие ошибки') tts.speak(errors_alarm)
И теперь нам осталось добавить в основной скрипт функции записи события в календарь – schedule_list() и вывода списка событий – schedule_add(), которые вызываются в command_processing():
def schedule_add(phrase: str): # Создаем экземпляр парсера parser = reminder_parser.SmartReminderParser() result = parser.parse(phrase) if result: print(f"📝 Фраза: {result['original']}") print(f"⏰ Время: {result['time']}") print(f"📋 Действие: '{result['text']}'") print(f"🔧 Тип: {result['time_type']}") print(f"📊 Timestamp: {result['timestamp']}") print("-" * 70) dt = result['time'] start = dt.isoformat() # Преобразуем в формат времени end = (dt + datetime.timedelta(hours=1)).isoformat() # Будем считать, что событите будет длиться час descr = "📝 Задание отправлено с умной колонки" event = result['text'] try: text = schedule.create_event(event.capitalize(), start, end, description=descr) # Создаем событие в календаре tts.speak(text) # Говорим, что событие успешно создано return False except: tts.speak("Извините, возникла ошибка записи события в календарь. Попробуйте еще раз.") return True else: tts.speak("Я не смогла распознать событие, пожалуйста, попробуйте еще раз.") return True
Где метод create_event() отвечает за создание события в Google календаре. Результат работы данной функции можно также наблюдать в терминале в процессе отладки:
Ниже на видео представлена работа функции добавления события в календарь:
Youtube
Rutube
После этого мы можем видеть наше событие в Google календаре:
И функция озвучивания предстоящих событий:
def schedule_list(): """Озвучиваем список предстоящих событий""" try: events = schedule.get_events_list() if not events: print('Нет предстоящих событий.') tts.speak("Нет предстоящих событий.") else: print(events) tts.speak("Нашла следующие события.") for even in events: tts.speak(" " + even['summary']) start = even['start'].get('dateTime', even['start'].get('date')) print(str(start)) dt = datetime.datetime.fromisoformat(start) text = "Запланировано на" male_units = ((u'час', u'часа', u'часов'), 'm') text += num2words_ru.num2text(dt.hour, male_units) + '.' male_units = ((u'минута', u'минуты', u'минут'), 'f') text += num2words_ru.num2text(dt.minute, male_units) + '.' tts.speak(text) except: tts.speak("Сожалею, но возникла ошибка, попробуйте позже!")
Ниже на видео вы можете наблюдать работу данной функции:
Youtube
Rutube
❯ Итоги
Пока на этом можно и закончить статью. Спасибо, что дочитали :). «А где оповещение умной колонки о наступающих событиях?» – скажете вы, да оно есть, но это уже контент для следующей статьи.
Если у вас есть вопросы, пожелания или советы – добро пожаловать в комментарии! Интересных проектов и спасибо за внимание!
Ссылки к статье:
Новости, обзоры продуктов и конкурсы от команды Timeweb.Cloud - в нашем Telegram-канале ↩
