jarvis-server/jarvis/skills/__init__.py

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, client_utils, config_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):
        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:
                # TODO: replace when found a better TTS engine for french
                # as the french tts don't support float in sentence, convert it to an integer
                if is_float(val):
                    # val = str(int(float(val)))  # convert a float to integer
                    val = str(val).split(".")[0] + " virgule " + str(val).split(".")[1]

                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_entities_padatious()
            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_entities_padatious(self):
        path = self.path + "/vocab/" + languages_utils.get_language() + "/*.entity"

        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:
            intent_manager.register_entity_padatious(filename, all_lines_by_file_dict.get(filename))

    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:
                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)


def is_float(value):
    try:
        float(value)
        return True
    except ValueError:
        return False


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__']]
                        elif intent_type == 'padatious':
                            intent_file = properties[0]
                            intent_name = properties[1]

                            intent_category = str(attrs['__module__']).split('.')[2]
                            skill_folder = str(attrs['__module__']).split('.')[3]

                            intent_manager.intents_handlers_padatious[f"{intent_name}"] = [key,
                                                                                           get_array_for_intent_file(
                                                                                               intent_file,
                                                                                               intent_category,
                                                                                               skill_folder),
                                                                                           attrs['__module__']]