add HomeAssistant get info on entity

Reviewed-on: #3

.gitignore

@@ -6,8 +6,8 @@ __pycache__/
 *$py.class
 
 # Custom
-model_small/
+data/model_small/
-model_large/
+data/model_large/
 
 # C extensions
 *.so

config.py

@@ -1,5 +0,0 @@
-VA_ALIAS = ('джарвис',)
-VA_TBR = ('скажи', 'покажи', 'ответь', 'произнеси', 'расскажи', 'сколько', 'слушай')
-MICROPHONE_INDEX = -1
-PICOVOICE_TOKEN = "4xbwaZwZmSHeTiowFl5Rgqsc8CR4FKGV8YueJUlR4Zt2e1kB64IDcA=="
-OPENAI_TOKEN = "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"

data/commands.yaml

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

data/config.py

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

data/home_assistant_entities.yaml

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

main.py

@@ -1,4 +1,4 @@
-from Jarvis import Jarvis
+from modules.Jarvis import Jarvis
 
 
 def main():

modules/HomeAssistant.py

@@ -0,0 +1,67 @@
+import requests
+import yaml
+from fuzzywuzzy import process
+
+from data import config
+
+
+class HomeAssistant:
+    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):
+        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="выключи телевизор"):
+        response = requests.post(
+            url=f"{self.url}/services/conversation/process",
+            json={"text": command},
+            headers={
+                "Authorization": "Bearer " + self.token,
+                "content-type": "application/json"
+            },
+        )
+        if response.status_code == 200:
+            return True
+        return False
+
+    def voice_to_name(self, voice: str) -> 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):
+        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", None)
+            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

@@ -10,21 +10,24 @@ import yaml
 from fuzzywuzzy import fuzz
 from pvrecorder import PvRecorder
 
-import config
+from data import config
-from utils import execute_cmd, play
+from modules import HomeAssistant
+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('commands.yaml', encoding='utf8'))
+        self.VA_CMD_LIST = yaml.safe_load(open('data/commands.yaml', encoding='utf8'))
+        self.home_assistant = HomeAssistant.HomeAssistant()
         self.porcupine = pvporcupine.create(
             access_key=config.PICOVOICE_TOKEN,
             keywords=['jarvis'],
             sensitivities=[1]
         )
-        self.kaldi_rec = vosk.KaldiRecognizer(vosk.Model("model_large"), 16000)
+        self.kaldi_rec = vosk.KaldiRecognizer(vosk.Model("data/model_small"), 16000)
 
     def main(self):
         self.recorder = PvRecorder(
@@ -70,6 +73,7 @@ class Jarvis:
                 if vrt > rc['percent']:
                     rc['cmd'] = c
                     rc['percent'] = vrt
+                    rc['recognized_phrase'] = x
         if len(rc['cmd'].strip()) <= 0:
             return False
         elif rc['percent'] < 70 or rc['cmd'] not in self.VA_CMD_LIST.keys():
@@ -77,7 +81,7 @@ class Jarvis:
             time.sleep(1)
             return False
         else:
-            execute_cmd.execute_cmd(self, rc['cmd'])
+            execute_cmd.execute_cmd(self, rc['cmd'], rc['recognized_phrase'], voice)
             return True
 
     def play(self, phrase, wait_done=True):

utils/download_models.py

@@ -0,0 +1,26 @@
+import os
+import sys
+
+from data import config
+
+
+def install_vosk_model():
+    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")
+        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")
+        elif sys.platform == "win32":
+            os.system(f"curl https://alphacephei.com/vosk/models/{config.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("del /s /q 1.zip")

utils/execute_cmd.py

@@ -1,4 +1,4 @@
-def execute_cmd(self, cmd: str):
+def execute_cmd(self, cmd: str, recognized_phrase: str, voice: str):
     if cmd == 'thanks':
         self.play("thanks")
     elif cmd == 'stupid':
@@ -7,3 +7,9 @@ def execute_cmd(self, cmd: str):
         self.play("off", True)
         self.porcupine.delete()
         exit(0)
+    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

@@ -15,7 +15,7 @@ def play(self, phrase, wait_done=True):
     if wait_done:
         self.recorder.stop()
     if filename:
-        wave_obj = sa.WaveObject.from_wave_file(f"{self.CDIR}/sound/{filename}")
+        wave_obj = sa.WaveObject.from_wave_file(f"{self.CDIR}/data/sound/{filename}")
         play_obj = wave_obj.play()
         if wait_done:
             play_obj.wait_done()