Привет, Хабр!
Время от времени я возвращаюсь к своему 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()Ниже видео живого теста парсера:
На видео показан промежуточный тест парсера, где в терминале отображается результат парсинга, а озвучивание произнесенной фразы выполняется для удобства, чтобы убедиться в корректности транскрибации в период отладки.
❯ Финальная интеграция
Если вы дочитали до этого момента, то поздравляю, мы близки к финалу :). Итак, давайте для понимания резюмируем то, что мы сделали выше. Мы написали два программных модуля, которые отвечают за работу с сервисом 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 календаре. Результат работы данной функции можно также наблюдать в терминале в процессе отладки:
Ниже на видео представлена работа функции добавления события в календарь:
После этого мы можем видеть наше событие в 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("Сожалею, но возникла ошибка, попробуйте позже!")Ниже на видео вы можете наблюдать работу данной функции:
❯ Итоги
Пока на этом можно и закончить статью. Спасибо, что дочитали :). «А где оповещение умной колонки о наступающих событиях?» – скажете вы, да оно есть, но это уже контент для следующей статьи.
Если у вас есть вопросы, пожелания или советы – добро пожаловать в комментарии! Интересных проектов и спасибо за внимание!
Ссылки к статье:
Новости, обзоры продуктов и конкурсы от команды Timeweb.Cloud - в нашем Telegram-канале ↩