started re-implementing old intent system

.idea/jarvis-server-v2.iml

@@ -10,6 +10,11 @@
     <orderEntry type="jdk" jdkName="Python 3.10 (jarvis-server-v2)" jdkType="Python SDK" />
     <orderEntry type="sourceFolder" forTests="false" />
   </component>
+  <component name="PackageRequirementsSettings">
+    <option name="versionSpecifier" value="Greater or equal (&gt;=x.y.z)" />
+    <option name="removeUnused" value="true" />
+    <option name="keepMatchingSpecifier" value="false" />
+  </component>
   <component name="TemplatesService">
     <option name="TEMPLATE_CONFIGURATION" value="Jinja2" />
   </component>

jarvis/api.py

@@ -4,6 +4,8 @@ import tempfile
 import requests
 from flask import request, Flask
 
+from jarvis.skills import intent_manager
+
 app = Flask(__name__)
 
 
@@ -16,7 +18,22 @@ def process_audio_request_android():
     audio_temp_file.write(request.data)
     print(audio_temp_file.name)
 
-    return {"transcription": text_recognition_whisperasr(audio_temp_file.name), "answer": "WIP"}
+    text = text_recognition_whisperasr(audio_temp_file.name)
+
+    # TODO: send to each skill to answer the questions
+
+    return {"transcription": text, "answer": "I'm still learning how to respond to that..."}
+
+
+@app.route("/process_text", methods=['POST'])
+def process_text():
+    print("[" + request.remote_addr + "] - New TXT request")
+
+    text = request.values['text']
+
+    answer = intent_manager.recognise(text, request.headers.get('Client-Ip'), request.headers.get('Client-Port'))
+
+    return {"transcription": text, "answer": answer}
 
 
 # send request to whisper-asr server (docker)

jarvis/skills/__init__.py

@@ -0,0 +1,159 @@
+import glob
+import os
+import random
+import threading
+import types
+
+from .. import get_path_file
+from ..skills import intent_manager
+from ..utils import languages_utils
+
+
+class Skill:
+    def __init__(self, name, data, required_config: list = None):
+        if required_config is not None:
+            self.required_config = required_config
+
+        self.name = name
+
+        self.client_ip = data['client_ip']
+        self.client_port = data['client_port']
+
+        path = self.__module__.split(".")
+        self.category = path[2]
+        self.skill_folder = path[3]
+
+        self.path = os.path.dirname(get_path_file.__file__) + "/skills/" + self.category + "/" + self.skill_folder
+
+        if self.has_required_config():
+            self.on_load()
+
+    def on_load(self):
+        pass
+
+    def speak(self, sentence):
+        # TODO: implement
+        print(sentence)
+        # client_utils.speak(sentence, self.client_ip, self.client_port)
+
+    def speak_dialog(self, dialog, data=None):
+        if data is None:
+            data = {}
+
+        file = self.path + "/dialog/" + languages_utils.get_language() + "/" + dialog + ".dialog"
+        random_line = get_random_line_from_file(file)
+
+        for key, val in data.items():
+            if "{{" + key + "}}" in random_line or "{" + key + "}" in random_line:
+                random_line = random_line.replace("{{" + key + "}}", val)
+                random_line = random_line.replace("{" + key + "}", val)
+
+        self.speak(random_line)
+
+        return "Error, dialog not found for : " + dialog
+
+    def speak_dialog_threaded(self, dialog, data=None):
+        thread = threading.Thread(target=self.speak_dialog, args=[dialog, data])
+        thread.start()
+
+    def register(self):
+        if self.has_required_config():
+            self.register_entities_adapt()
+            self.register_regex()
+            print("[" + self.name + "] Registered entity/entities and regex(s)")
+        else:
+            print("[WARN] Skipped skill " + self.name + " : please check your config.")
+
+    def register_entities_adapt(self):
+        path = self.path + "/vocab/" + languages_utils.get_language() + "/*.voc"
+
+        all_lines_by_file_dict = get_lines_of_all_files_in_path(path, return_as_dict_with_filename=True)
+
+        for filename in all_lines_by_file_dict:
+            for line in all_lines_by_file_dict.get(filename):
+                intent_manager.register_entity_adapt(line, filename, self.name)
+
+    def register_regex(self):
+        path = self.path + "/regex/" + languages_utils.get_language() + "/*.rx"
+
+        result = get_lines_of_all_files_in_path(path)
+        for line in result:
+            intent_manager.register_regex_adapt(line, self.name)
+
+    def has_required_config(self):
+        if hasattr(self, "required_config") and self.required_config is not None:
+            for field in self.required_config:
+                # TODO: implement
+                # if config_utils.get_in_config(field) is None:
+                    return False
+        return True
+
+
+def get_random_line_from_file(filepath):
+    if os.path.exists(filepath):
+        with open(filepath, "r") as infile:
+            random_line = random.choice(infile.readlines())
+            infile.close()
+            return random_line
+    else:
+        print("File " + filepath + " doesn't exist...")
+
+
+def get_all_lines_from_file(filepath):
+    if os.path.exists(filepath):
+        with open(file=filepath, mode="r") as infile:
+            lines = []
+
+            for line in infile.readlines():
+                lines.append(line.removesuffix('\n'))
+
+            infile.close()
+
+            return lines
+    else:
+        print("File " + filepath + " doesn't exist...")
+
+
+def get_lines_of_all_files_in_path(path, return_as_dict_with_filename=False):
+    files = glob.glob(path, recursive=True)
+    result = dict()
+    result_list = []
+
+    for file in files:
+        lines = get_all_lines_from_file(file)
+
+        if return_as_dict_with_filename:
+            filename = file.split("/")[-1].split('.')[0]
+            result[filename] = lines
+        else:
+            result_list.extend(lines)
+
+    if return_as_dict_with_filename:
+        return result
+
+    return result_list
+
+
+def get_array_for_intent_file(filename, category, skill_folder):
+    path = os.path.dirname(get_path_file.__file__) + "/skills/" + category + "/" + skill_folder
+    path = path + "/vocab/" + languages_utils.get_language() + "/" + filename
+
+    return get_all_lines_from_file(path)
+
+
+class SkillRegistering(type):
+    def __init__(cls, name, bases, attrs):
+        for key, val in attrs.items():
+            if type(val) is types.FunctionType and not str(val).__contains__("__"):
+                intent_type = getattr(val, "_type", None)
+
+                if intent_type is not None:
+                    properties = getattr(val, "_data", None)
+
+                    if properties is not None:
+                        if intent_type == 'adapt':
+                            intent = properties[0]
+                            intent_name = intent.name
+
+                            intent_manager.intents_handlers_adapt[f"{intent_name}"] = [getattr(cls, key), name, key,
+                                                                                       attrs['__module__']]

jarvis/skills/decorators.py

@@ -0,0 +1,26 @@
+def intent_handler(*args):
+    """
+    Creates an attribute on the method, so it can
+    be discovered by the metaclass
+    """
+
+    def decorator(f):
+        f._type = "adapt"
+        f._data = args
+        return f
+
+    return decorator
+
+
+def intent_file_handler(*args):
+    """
+    Creates an attribute on the method, so it can
+    be discovered by the metaclass
+    """
+
+    def decorator(f):
+        f._type = "padatious"
+        f._data = args
+        return f
+
+    return decorator

jarvis/skills/entertainement/jokes/__init__.py

@@ -0,0 +1,38 @@
+from jarvis.skills import Skill, SkillRegistering
+from jarvis.skills.decorators import intent_file_handler
+from jarvis.skills.entertainement.jokes.providers import jokes_french_provider, jokes_english_provider
+from jarvis.utils import languages_utils
+
+
+class JokesSkill(Skill, metaclass=SkillRegistering):
+    def __init__(self, data=dict):
+        super().__init__("JokesSkill", data)
+
+    @intent_file_handler("tell_a_joke.intent", "TellAJokeIntent")
+    def handle_joke(self, data):
+        self.speak(get_joke(False))
+
+
+def get_joke(nsfw, lang=languages_utils.get_language()):
+    """
+    Returns a joke in the good language
+
+    Args:
+        lang: use language from config by default, you can specify a custom language here
+        nsfw: should include nsfw jokes
+
+    Returns:
+        array
+    """
+    return "really nice joke"
+
+    if lang.startswith("fr-"):
+        return jokes_french_provider.get_joke(nsfw)
+    elif lang.startswith("en-"):
+        return jokes_english_provider.get_joke(nsfw)
+    else:
+        return ['Error', "I don't know any jokes in your language..."]
+
+
+def create_skill(data):
+    return JokesSkill(data)

jarvis/skills/entertainement/jokes/providers/jokes_english_provider.py

@@ -0,0 +1,25 @@
+import requests
+
+
+def get_joke(nsfw=False):
+    """
+    Returns a joke in 2 parts
+
+    Args:
+        nsfw: include nsfw jokes?
+
+    Returns: array
+    """
+
+    url = 'https://v2.jokeapi.dev/joke/Any?blacklistFlags=nsfw,religious,political,racist,sexist,explicit&type=twopart'
+
+    if nsfw:
+        url = "https://v2.jokeapi.dev/joke/Any?type=twopart"
+
+    response = requests.get(url)
+
+    data = response.json()
+    joke = data['setup']
+    answer = data['delivery']
+
+    return [joke, answer]

jarvis/skills/entertainement/jokes/providers/jokes_french_provider.py

@@ -0,0 +1,27 @@
+import requests
+
+
+def get_joke(nsfw=False):
+    """
+    Returns a joke in 2 parts
+
+    Args:
+        nsfw: include nsfw jokes?
+
+    Returns: array
+    """
+
+    url = 'https://www.blagues-api.fr/api/random'
+
+    if nsfw:
+        url = url + "?disallow=dark&disallow=limit"
+
+    # please register on www.blagues-api.fr and set a token in your secrets file
+    response = requests.get(url, headers={
+        'Authorization': 'Bearer ' + """ config_utils.get_in_secret('JOKES_FRENCH_API_TOKEN') """})
+
+    data = response.json()
+    joke = data['joke']
+    answer = data['answer']
+
+    return [joke, answer]

jarvis/skills/entertainement/jokes/vocab/fr-fr/tell_a_joke.intent

@@ -0,0 +1,3 @@
+raconte (moi|nous|) (voir|) une blague
+fais (moi|nous) rire
+illumine (ma|notre) journée

jarvis/skills/intent_manager.py

@@ -0,0 +1,119 @@
+import importlib
+
+from adapt.engine import DomainIntentDeterminationEngine
+
+adapt_engine = DomainIntentDeterminationEngine()
+
+intents_handlers_adapt = dict()
+intents_handlers_padatious = dict()
+
+
+def register_entity_adapt(entity_value, entity_type, domain):
+    adapt_engine.register_entity(entity_value=entity_value, entity_type=entity_type, domain=domain)
+    # print("[Adapt]: Added entity with type " + entity_type + " for " + domain)
+
+
+def register_regex_adapt(regex, domain):
+    adapt_engine.register_regex_entity(regex, domain)
+    # print("[Adapt]: Added new regex for " + domain)
+
+
+def register_intent_adapt(intent, domain):
+    adapt_engine.register_intent_parser(intent, domain=domain)
+    print("[Adapt]: Registered new intent " + intent.name + " for skill " + domain + ".")
+
+
+def load_all_skills():
+    for handler in intents_handlers_adapt:
+        function_handler = intents_handlers_adapt.get(handler)
+        intent_builder = getattr(function_handler[0], "_data", [])[0]
+        skill_name = function_handler[1]
+        register_intent_adapt(intent_builder.build(), domain=skill_name)
+
+
+def handle(intent_name, data):
+    module_path_str = None
+    handler_method_name = None
+
+    if intent_name in intents_handlers_adapt:
+        handler_method_name = intents_handlers_adapt.get(intent_name)[2]
+        module_path_str = intents_handlers_adapt.get(intent_name)[3]
+
+    if module_path_str is not None and handler_method_name is not None:
+        # import the create_skill method from the skill using the skill module path
+        create_skill_method = import_method_from_string(module_path_str, "create_skill")
+
+        skill_init_data = {'client_ip': data['client_ip'], 'client_port': data['client_port']}
+
+        # create a new object of the right skill for the utterance
+        skill = create_skill_method(skill_init_data)
+
+        # import and call the handler method from the skill
+        getattr(skill, handler_method_name)(data=data)
+
+
+def import_method_from_string(file, method_name):
+    """
+    Add the possibility to import method dynamically using a string like "skills.daily.date_and_time.intent" as file and
+    "what_time_is_it" as method_name
+    """
+    mod = importlib.import_module(file)
+    met = getattr(mod, method_name)
+
+    return met
+
+
+def recognise(sentence, client_ip=None, client_port=None):
+    sentence = sentence.lower()
+    print(sentence)
+
+    data = dict()
+    data['client_ip'] = client_ip
+    data['client_port'] = client_port
+    data['utterance'] = sentence
+
+    best_intent_adapt = get_best_intent_adapt(sentence)
+
+    confidence_adapt = get_confidence(best_intent_adapt)
+
+    if confidence_adapt < 0.2:
+
+        # TODO: implement fallback to something like wolfram alpha or smart assistant
+        # Nothing found at all
+        return "I didn't understand..."
+    else:
+        return handle_adapt_intent(data, best_intent_adapt)
+
+
+def get_confidence(intent):
+    if intent is None:
+        return 0
+
+    if 'confidence' in intent:
+        return intent['confidence']
+    elif hasattr(intent, 'conf'):
+        return intent.conf
+    else:
+        return 0
+
+
+def get_best_intent_adapt(sentence):
+    if len(intents_handlers_adapt) > 0:
+        try:
+            best_intents = adapt_engine.determine_intent(sentence, 100)
+            best_intent = next(best_intents)
+
+            return best_intent
+
+        except StopIteration:
+            pass
+
+    return None  # No match (Adapt)
+
+
+def handle_adapt_intent(data, intent):
+    for key, val in intent.items():
+        if key != 'intent_type' and key != 'target' and key != 'confidence':
+            data[key] = val
+    handle(intent['intent_type'], data=data)
+    return intent

jarvis/start.py

@@ -1,4 +1,15 @@
 import api
+import lingua_franca
+
+from jarvis.skills.entertainement.jokes import JokesSkill
 
 if __name__ == '__main__':
+
+    # Load lingua franca in the memory
+    lingua_franca.load_language(lang="fr")
+
+    # Load skills
+    JokesSkill().register()
+
+    # Start the api endpoint
     api.start_server()

jarvis/utils/languages_utils.py

@@ -0,0 +1,2 @@
+def get_language():
+    return "fr"

requirements.txt

@@ -1,2 +1,4 @@
 requests~=2.28.1
 Flask~=2.2.2
+adapt-parser==1.0.0
+lingua-franca~=0.4.3