add comment

Reviewed-on: #5

.env.example

@@ -0,0 +1,10 @@
+VA_ALIAS='("джарвис",)'
+VA_TBR='("скажи", "покажи", "ответь", "произнеси", "расскажиv, "сколько", "слушай")'
+VOSK_MODEL_NAME='vosk-model-small-ru-0.22'  # vosk-model-ru-0.42
+MICROPHONE_INDEX=-1
+PICOVOICE_TOKEN='token'
+
+
+# home assistant
+HOME_ASSISTANT_URL='http://localhost:8123/api'
+HOME_ASSISTANT_TOKEN=''

data/commands.yaml

@@ -72,7 +72,9 @@ weather:
   - возможен дождь сегодня?
   - прогноз погоды на сегодня
   - погода
-home_assistant:
+home_assistant_execute:
   - включи телевизор
   - выключи телевизор
-  - начни уборку
+  - начни уборку
+home_assistant_get:
+  - тест

data/config.py

@@ -1,10 +1,15 @@
-VA_ALIAS = ("джарвис",)
-VA_TBR = ("скажи", "покажи", "ответь", "произнеси", "расскажи", "сколько", "слушай")
-MODEL_NAME = "vosk-model-small-ru-0.22"  # vosk-model-ru-0.42
-MICROPHONE_INDEX = -1
-PICOVOICE_TOKEN = "4xbwaZwZmSHeTiowFl5Rgqsc8CR4FKGV8YueJUlR4Zt2e1kB64IDcA=="
+import environs
 
+env = environs.Env()
+env.read_env()
+
+
+VA_ALIAS = env.str("VA_ALIAS")
+VA_TBR = env.str("VA_TBR")
+VOSK_MODEL_NAME = env.str("VOSK_MODEL_NAME")
+MICROPHONE_INDEX = env.int("MICROPHONE_INDEX")
+PICOVOICE_TOKEN = env.str("PICOVOICE_TOKEN")
 
 # home assistant
-HOME_ASSISTANT_URL = "http://192.168.0.112:9999/api"
-HOME_ASSISTANT_TOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiI5NjczNDZjYjc2YzI0YWQzODdhMmUwMmM2MjViZGVjZCIsImlhdCI6MTcxNDQ3MzkzNywiZXhwIjoyMDI5ODMzOTM3fQ.TATpIMXivJOioCtUI8PKg6gyTQYMG6bur6enm6NxjtY"
+HOME_ASSISTANT_URL = env.str("HOME_ASSISTANT_URL")
+HOME_ASSISTANT_TOKEN = env.str("HOME_ASSISTANT_TOKEN")

data/home_assistant_entities.yaml

@@ -0,0 +1,4 @@
+пылесос:
+  - entity_id:vacuum.roborock_vacuum_m1s
+  - state:находится в
+  - attributes.battery_level:а его уровень зарядки

modules/HomeAssistant.py

@@ -1,23 +1,45 @@
 import requests
+import yaml
+from fuzzywuzzy import process
+from requests import Response
 
 from data import config
 
 
 class HomeAssistant:
+    """
+    Модуль home assistant для работы с его api
+    """
     def __init__(self):
         self.url = "http://192.168.0.112:9999/api"
         self.token = config.HOME_ASSISTANT_TOKEN
+        self.HA_CMD_LIST = yaml.safe_load(open('data/home_assistant_entities.yaml', encoding='utf8'))
 
-    def get_info(self, state):
+    def get_info(self, state: str) -> Response:
+        """
+        Функция для получения информации о заданном entity
+
+        :param state: str - объект в home assistant информацию о котором надо узнать
+        :return: Response - ответ от сервера api
+        """
         response = requests.get(
             url=f"{self.url}/states",
             headers={
                 "Authorization": "Bearer " + self.token
             }
         )
+        for entity in response.json():
+            if entity["entity_id"] == state:
+                return entity
         return response
 
-    def send_process(self, command="выключи телевизор"):
+    def send_process(self, command: str = "выключи телевизор") -> bool:
+        """
+        Функция для отправки запроса о выполнении команды к api
+
+        :param command: str - команда в виде строки
+        :return: bool - удачная ли отправка запроса к api
+        """
         response = requests.post(
             url=f"{self.url}/services/conversation/process",
             json={"text": command},
@@ -29,3 +51,47 @@ class HomeAssistant:
         if response.status_code == 200:
             return True
         return False
+
+    def voice_to_name(self, voice: str) -> str:
+        """
+        Функция для неточного сравнивания входной строки голоса
+        и списка устройств дял которых можно узнать информацию
+
+        :param voice: str - распознанная фраза без проверки по списку
+        :return: str - найденный объект для получения информации
+        """
+        words = voice.lower().split()
+        best_match = None
+        highest_score = 0
+        for word in words:
+            result, score = process.extractOne(word, self.HA_CMD_LIST.keys())
+            if score > highest_score:
+                highest_score = score
+                best_match = result
+        return best_match
+
+    def validate_info(self, name: str) -> str:
+        """
+        Функция для получения готовой строки информации entity по его имени.
+        Эта строка готова для произношения
+
+        :param name: str - имя entity для нахождения информации о нём
+        :return: str - готовая строка для найденного по имени объекта для её произношения
+        """
+        answer = name
+        entity_config = self.HA_CMD_LIST.get(name)
+        if entity_config:
+            # Создание словаря, разделяя каждый элемент конфигурации на ключ и значение
+            entity_details = {item.split(':')[0]: item.split(':')[1] for item in entity_config}
+            entity_id = entity_details.pop("entity_id", "robot")
+            if entity_id:
+                responses = self.get_info(entity_id)
+                for attribute_path, label in entity_details.items():
+                    response = responses
+                    try:
+                        for attribute in attribute_path.split("."):
+                            response = response[attribute]
+                        answer += f" {label} {response}"
+                    except KeyError:
+                        continue
+        return answer

modules/Jarvis.py

@@ -11,17 +11,21 @@ from fuzzywuzzy import fuzz
 from pvrecorder import PvRecorder
 
 from data import config
-from modules.HomeAssistant import HomeAssistant
+from modules import HomeAssistant, MediaPlayerController
 from utils import download_models, execute_cmd, play
 
 
 class Jarvis:
+    """
+    Это основной модуль голосового ассистента
+    """
     def __init__(self):
         download_models.install_vosk_model()
         self.recorder = None
         self.CDIR = os.getcwd()
         self.VA_CMD_LIST = yaml.safe_load(open('data/commands.yaml', encoding='utf8'))
-        self.home_assistant = HomeAssistant()
+        self.home_assistant = HomeAssistant.HomeAssistant()
+        self.media_player_controller = MediaPlayerController.MediaPlayerController()
         self.porcupine = pvporcupine.create(
             access_key=config.PICOVOICE_TOKEN,
             keywords=['jarvis'],
@@ -62,9 +66,15 @@ class Jarvis:
                 print(f"Unexpected {err=}, {type(err)=}")
                 raise
 
-    def va_respond(self, voice: str):
+    def va_respond(self, voice: str) -> bool:
+        """
+        Функция предсказывает команду
+
+        :param voice: str - распознанная строка
+        :return: bool - распознана или нет команда
+        """
         print(f"Распознано: {voice}")
-        for x in config.VA_ALIAS + config.VA_TBR:
+        for x in config.VA_ALIAS:
             voice = voice.replace(x, "").strip()
         rc = {'cmd': '', 'percent': 0}
         for c, v in self.VA_CMD_LIST.items():
@@ -84,5 +94,13 @@ class Jarvis:
             execute_cmd.execute_cmd(self, rc['cmd'], rc['recognized_phrase'], voice)
             return True
 
-    def play(self, phrase, wait_done=True):
+    def play(self, phrase: str, wait_done: bool = True):
+        """
+        Функция для запуска голосовой команды
+
+        :param self: modules.Jarvis - объект основного модуля
+        :param phrase: str - фраза для запуска голосовой команды
+        :param wait_done: bool - нужно-ли ждать окончания фразы
+        :return:
+        """
         play.play(self, phrase, wait_done)

modules/MediaPlayerController.py

@@ -0,0 +1,90 @@
+import platform
+import subprocess
+
+
+class MediaPlayerController:
+    """
+    Модуль для манипуляции музыкой
+    """
+    def __init__(self):
+        self.os_type = platform.system()
+
+    def play_pause(self) -> None:
+        """
+        Запуск/остановка музыки
+
+        :return:
+        """
+        if self.os_type == 'Windows':
+            self._windows_play_pause()
+        elif self.os_type == 'Linux':
+            self._linux_control("play-pause")
+
+    def next_track(self) -> None:
+        """
+        Включает следующею композицию
+
+        :return:
+        """
+        if self.os_type == 'Windows':
+            self._windows_control("next")
+        elif self.os_type == 'Linux':
+            self._linux_control("next")
+
+    def previous_track(self) -> None:
+        """
+        Включает предыдущею композицию
+
+        :return:
+        """
+        if self.os_type == 'Windows':
+            self._windows_control("previous")
+        elif self.os_type == 'Linux':
+            self._linux_control("previous")
+
+    def _windows_play_pause(self) -> None:
+        """
+        Запуск/остановка музыки в windows
+
+        :return:
+        """
+        import win32con
+        self.key_press(win32con.VK_MEDIA_PLAY_PAUSE)
+
+    def _windows_control(self, action: str) -> None:
+        """
+        Включает предыдущею или следующею композицию в windows
+
+        :return:
+        """
+        import win32con
+        if action == "next":
+            self.key_press(win32con.VK_MEDIA_NEXT_TRACK)
+        elif action == "previous":
+            self.key_press(win32con.VK_MEDIA_PREV_TRACK)
+
+    @staticmethod
+    def key_press(key_code: str) -> None:
+        """
+        Симуляция нажатия и отпускания клавиши
+
+        :param key_code: str - какую кнопку нажать
+        :return:
+        """
+        import win32api
+        import win32con
+        win32api.keybd_event(key_code, 0, 0, 0)
+        win32api.keybd_event(key_code, 0, win32con.KEYEVENTF_KEYUP, 0)
+
+    @staticmethod
+    def _linux_control(command: str) -> None:
+        """
+        Запускает команду для linux систем
+
+        :param command: str - команда для запуска
+        :return:
+        """
+        try:
+            subprocess.run(["playerctl", command], check=True)
+        except subprocess.CalledProcessError as e:
+            print(f"Failed to {command}: {e}")

poetry.lock

@@ -329,6 +329,26 @@ files = [
     {file = "decorator-5.1.1.tar.gz", hash = "sha256:637996211036b6385ef91435e4fae22989472f9d571faba8927ba8253acbc330"},
 ]
 
+[[package]]
+name = "environs"
+version = "11.0.0"
+description = "simplified environment variable parsing"
+optional = false
+python-versions = ">=3.8"
+files = [
+    {file = "environs-11.0.0-py3-none-any.whl", hash = "sha256:e0bcfd41c718c07a7db422f9109e490746450da38793fe4ee197f397b9343435"},
+    {file = "environs-11.0.0.tar.gz", hash = "sha256:069727a8f73d8ba8d033d3cd95c0da231d44f38f1da773bf076cef168d312ee8"},
+]
+
+[package.dependencies]
+marshmallow = ">=3.13.0"
+python-dotenv = "*"
+
+[package.extras]
+dev = ["environs[tests]", "pre-commit (>=3.5,<4.0)", "tox"]
+django = ["dj-database-url", "dj-email-url", "django-cache-url"]
+tests = ["environs[django]", "pytest"]
+
 [[package]]
 name = "filelock"
 version = "3.13.4"
@@ -974,6 +994,25 @@ files = [
     {file = "MarkupSafe-2.1.5.tar.gz", hash = "sha256:d283d37a890ba4c1ae73ffadf8046435c76e7bc2247bbb63c00bd1a709c6544b"},
 ]
 
+[[package]]
+name = "marshmallow"
+version = "3.21.2"
+description = "A lightweight library for converting complex datatypes to and from native Python datatypes."
+optional = false
+python-versions = ">=3.8"
+files = [
+    {file = "marshmallow-3.21.2-py3-none-any.whl", hash = "sha256:70b54a6282f4704d12c0a41599682c5c5450e843b9ec406308653b47c59648a1"},
+    {file = "marshmallow-3.21.2.tar.gz", hash = "sha256:82408deadd8b33d56338d2182d455db632c6313aa2af61916672146bb32edc56"},
+]
+
+[package.dependencies]
+packaging = ">=17.0"
+
+[package.extras]
+dev = ["marshmallow[tests]", "pre-commit (>=3.5,<4.0)", "tox"]
+docs = ["alabaster (==0.7.16)", "autodocsumm (==0.2.12)", "sphinx (==7.3.7)", "sphinx-issues (==4.1.0)", "sphinx-version-warning (==1.1.2)"]
+tests = ["pytest", "pytz", "simplejson"]
+
 [[package]]
 name = "matplotlib"
 version = "3.8.4"
@@ -2460,4 +2499,4 @@ files = [
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.11"
-content-hash = "80e7f3585f34cc051c27d91abbe3f4bff80e5ea9fa403103eeed5728b14c2cc5"
+content-hash = "f3829914195ce01bfac29dea5117fb3f28f095ecc6081a77eddb6bacc895718b"

pyproject.toml

@@ -27,6 +27,7 @@ torchaudio = "^2.1.1+cpu"
 ollama = "^0.1.6"
 ruff = "^0.4.2"
 noisereduce = "^3.0.2"
+environs = "^11.0.0"
 
 
 [[tool.poetry.source]]

test.py

@@ -1,5 +0,0 @@
-from modules.HomeAssistant import HomeAssistant
-
-home_assistant = HomeAssistant()
-response = home_assistant.get_info("")
-print(response.text)

utils/download_models.py

@@ -4,23 +4,34 @@ import sys
 from data import config
 
 
-def install_vosk_model():
+def install_vosk_model() -> None:
+    """
+    Функция устанавливает заданную в конфигурационном файле модели
+    
+    :return: 
+    """
     try:
         open('data/model_small/README')
     except Exception as e:
         print(e)
         if sys.platform == "linux" or sys.platform == "linux2":
-            os.system(f"wget https://alphacephei.com/vosk/models/{config.MODEL_NAME}.zip")
-            os.system(f"unzip {config.MODEL_NAME}.zip")
-            os.system(f"mv {config.MODEL_NAME} data/model_small")
-            os.system(f"rm -rf {config.MODEL_NAME}.zip")
+            os.system(
+                f"wget https://alphacephei.com/vosk/models/{config.VOSK_MODEL_NAME}.zip"
+            )
+            os.system(f"unzip {config.VOSK_MODEL_NAME}.zip")
+            os.system(f"mv {config.VOSK_MODEL_NAME} data/model_small")
+            os.system(f"rm -rf {config.VOSK_MODEL_NAME}.zip")
         elif sys.platform == "darwin":
-            os.system(f"curl https://alphacephei.com/vosk/models/{config.MODEL_NAME}.zip")
-            os.system(f"unzip {config.MODEL_NAME}.zip")
-            os.system(f"mv {config.MODEL_NAME} data/model_small")
-            os.system(f"rm -rf {config.MODEL_NAME}.zip")
+            os.system(
+                f"curl https://alphacephei.com/vosk/models/{config.VOSK_MODEL_NAME}.zip"
+            )
+            os.system(f"unzip {config.VOSK_MODEL_NAME}.zip")
+            os.system(f"mv {config.VOSK_MODEL_NAME} data/model_small")
+            os.system(f"rm -rf {config.VOSK_MODEL_NAME}.zip")
         elif sys.platform == "win32":
-            os.system(f"curl https://alphacephei.com/vosk/models/{config.MODEL_NAME}.zip --output 1.zip")
+            os.system(
+                f"curl https://alphacephei.com/vosk/models/{config.VOSK_MODEL_NAME}.zip --output 1.zip"
+            )
             os.system('powershell -command "Expand-Archive 1.zip ./"')
-            os.system(f"rename {config.MODEL_NAME} data/model_small")
+            os.system(f"rename {config.VOSK_MODEL_NAME} data/model_small")
             os.system("del /s /q 1.zip")

utils/execute_cmd.py

@@ -1,4 +1,13 @@
-def execute_cmd(self, cmd: str, recognized_phrase: str, voice: str):
+def execute_cmd(self, cmd: str, recognized_phrase: str, voice: str) -> None:
+    """
+    Функция выполняет полученные команды
+
+    :param self: modules.Jarvis - объект основного модуля
+    :param cmd: str - команда которую функция должна выполнить
+    :param recognized_phrase: str - распознанная фраза из списка фраз
+    :param voice: str - распознанная фраза без проверки по списку
+    :return:
+    """
     if cmd == 'thanks':
         self.play("thanks")
     elif cmd == 'stupid':
@@ -7,5 +16,17 @@ def execute_cmd(self, cmd: str, recognized_phrase: str, voice: str):
         self.play("off", True)
         self.porcupine.delete()
         exit(0)
-    elif cmd == 'home_assistant':
+    elif cmd == 'music_on':
+        self.media_player_controller.play_pause()
+    elif cmd == 'music_off':
+        self.media_player_controller.play_pause()
+    elif cmd == 'music_next':
+        self.media_player_controller.next_track()
+    elif cmd == 'music_previous':
+        self.media_player_controller.previous_track()
+    elif cmd == 'home_assistant_execute':
         self.home_assistant.send_process(recognized_phrase)
+    elif cmd == 'home_assistant_get':
+        entity_name = self.home_assistant.voice_to_name(voice)
+        entity_info = self.home_assistant.validate_info(entity_name)
+        print(entity_info)

utils/play.py

@@ -3,7 +3,15 @@ import random
 import simpleaudio as sa
 
 
-def play(self, phrase, wait_done=True):
+def play(self, phrase: str, wait_done: bool = True) -> None:
+    """
+    Функция для запуска голосовой команды
+
+    :param self: modules.Jarvis - объект основного модуля
+    :param phrase: str - фраза для запуска голосовой команды
+    :param wait_done: bool - нужно-ли ждать окончания фразы
+    :return:
+    """
     filename = None
     file_array = ["not_found", "thanks", "run", "stupid", "ready", "off"]
     if phrase == "greet":

utils/tts.py

@@ -1,31 +0,0 @@
-import time
-
-import sounddevice as sd
-import torch
-
-language = 'ru'
-model_id = 'ru_v3'
-sample_rate = 48000  # 48000
-speaker = 'aidar'  # aidar, baya, kseniya, xenia, random
-put_accent = True
-put_yo = True
-device = torch.device('cpu')  # cpu или gpu
-text = "Хауди Хо, друзья!!!"
-
-model, _ = torch.hub.load(repo_or_dir='snakers4/silero-models',
-                          model='silero_tts',
-                          language=language,
-                          speaker=model_id)
-model.to(device)
-
-
-def va_speak(what: str):
-    audio = model.apply_tts(text=what + "..",
-                            speaker=speaker,
-                            sample_rate=sample_rate,
-                            put_accent=put_accent,
-                            put_yo=put_yo)
-
-    sd.play(audio, sample_rate * 1.05)
-    time.sleep((len(audio) / sample_rate) + 0.5)
-    sd.stop()