192 lines
6.9 KiB
Python
192 lines
6.9 KiB
Python
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:
|
|
# 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)
|
|
|
|
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__']]
|