Utils.py 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303
  1. import logging
  2. import os
  3. import inspect
  4. import imp
  5. import sys
  6. import re
  7. import six
  8. logging.basicConfig()
  9. logger = logging.getLogger("kalliope")
  10. def pipe_print(line):
  11. line = Utils.encode_text_utf8(line)
  12. print(line)
  13. class ModuleNotFoundError(Exception):
  14. """
  15. The module can not been found
  16. .. notes: Check the case: must be in lower case.
  17. """
  18. pass
  19. class Utils(object):
  20. color_list = dict(
  21. PURPLE='\033[95m',
  22. BLUE='\033[94m',
  23. GREEN='\033[92m',
  24. YELLOW='\033[93m',
  25. RED='\033[91m',
  26. ENDLINE='\033[0m',
  27. BOLD='\033[1m',
  28. UNDERLINE='\033[4m'
  29. )
  30. ##################
  31. #
  32. # Shell properly displayed
  33. #
  34. #########
  35. @classmethod
  36. def print_info(cls, text_to_print):
  37. pipe_print(cls.color_list["BLUE"] + text_to_print + cls.color_list["ENDLINE"])
  38. logger.info(text_to_print)
  39. @classmethod
  40. def print_success(cls, text_to_print):
  41. pipe_print(cls.color_list["GREEN"] + text_to_print + cls.color_list["ENDLINE"])
  42. logger.info(text_to_print)
  43. @classmethod
  44. def print_warning(cls, text_to_print):
  45. pipe_print(cls.color_list["YELLOW"] + text_to_print + cls.color_list["ENDLINE"])
  46. logger.info(text_to_print)
  47. @classmethod
  48. def print_danger(cls, text_to_print):
  49. pipe_print(cls.color_list["RED"] + text_to_print + cls.color_list["ENDLINE"])
  50. logger.info(text_to_print)
  51. @classmethod
  52. def print_header(cls, text_to_print):
  53. pipe_print(cls.color_list["HEADER"] + text_to_print + cls.color_list["ENDLINE"])
  54. logger.info(text_to_print)
  55. @classmethod
  56. def print_purple(cls, text_to_print):
  57. pipe_print(cls.color_list["PURPLE"] + text_to_print + cls.color_list["ENDLINE"])
  58. logger.info(text_to_print)
  59. @classmethod
  60. def print_bold(cls, text_to_print):
  61. pipe_print(cls.color_list["BOLD"] + text_to_print + cls.color_list["ENDLINE"])
  62. logger.info(text_to_print)
  63. @classmethod
  64. def print_underline(cls, text_to_print):
  65. pipe_print(cls.color_list["UNDERLINE"] + text_to_print + cls.color_list["ENDLINE"])
  66. logger.info(text_to_print)
  67. @staticmethod
  68. def print_yaml_nicely(to_print):
  69. """
  70. Used for debug
  71. :param to_print: Dict to print nicely
  72. :return:
  73. """
  74. import json
  75. line = json.dumps(to_print, indent=2)
  76. return line.encode('utf-8')
  77. ##################
  78. #
  79. # Dynamic loading
  80. #
  81. #########
  82. @classmethod
  83. def get_dynamic_class_instantiation(cls, package_name, module_name, parameters=None, resources_dir=None):
  84. """
  85. Load a python class dynamically
  86. from my_package.my_module import my_class
  87. mod = __import__('my_package.my_module', fromlist=['my_class'])
  88. klass = getattr(mod, 'my_class')
  89. :param package_name: name of the package where we will find the module to load (neurons, tts, stt, trigger)
  90. :param module_name: name of the module from the package_name to load. This one is capitalized. Eg: Snowboy
  91. :param parameters: dict parameters to send as argument to the module
  92. :param resources_dir: the resource directory to check for external resources
  93. :return:
  94. """
  95. package_path = "kalliope." + package_name + "." + module_name.lower() + "." + module_name.lower()
  96. logger.debug("[Utils]-> get_dynamic_class_instantiation : package path : %s" % (package_path))
  97. if resources_dir is not None:
  98. neuron_resource_path = resources_dir + os.sep + module_name.lower() \
  99. + os.sep + module_name.lower() + ".py"
  100. if os.path.exists(neuron_resource_path):
  101. imp.load_source(module_name.capitalize(), neuron_resource_path)
  102. package_path = module_name.capitalize()
  103. logger.debug("[Utils]-> get_dynamic_class_instantiation : loading path : %s, as package %s" % (
  104. neuron_resource_path, package_path))
  105. mod = __import__(package_path, fromlist=[module_name.capitalize()])
  106. try:
  107. klass = getattr(mod, module_name.capitalize())
  108. except AttributeError:
  109. logger.debug("Error: No module named %s " % module_name.capitalize())
  110. raise ModuleNotFoundError(
  111. "The module %s does not exist in package %s" % (module_name.capitalize(), package_name))
  112. if klass is not None:
  113. # run the plugin
  114. if not parameters:
  115. return klass()
  116. elif isinstance(parameters, dict):
  117. return klass(**parameters)
  118. else:
  119. return klass(parameters)
  120. return None
  121. ##################
  122. #
  123. # Paths management
  124. #
  125. #########
  126. @staticmethod
  127. def get_current_file_parent_parent_path(current_script_path):
  128. parent_parent_path = os.path.normpath(current_script_path + os.sep + os.pardir + os.sep + os.pardir)
  129. return parent_parent_path
  130. @staticmethod
  131. def get_current_file_parent_path(current_script_path):
  132. parent_path = os.path.normpath(current_script_path + os.sep + os.pardir)
  133. return parent_path
  134. @classmethod
  135. def get_real_file_path(cls, file_path_to_test):
  136. """
  137. Try to return a full path from a given <file_path_to_test>
  138. If the path is an absolute on, we return it directly.
  139. If the path is relative, we try to get the full path in this order:
  140. - from the current directory where kalliope has been called + the file_path_to_test.
  141. Eg: /home/me/Documents/kalliope_config
  142. - from /etc/kalliope + file_path_to_test
  143. - from the default file passed as <file_name> at the root of the project
  144. :param file_path_to_test file path to test
  145. :type file_path_to_test: str
  146. :return: absolute path to the file file_path_to_test or None if is doen't exist
  147. """
  148. if not os.path.isabs(file_path_to_test):
  149. current_script_path = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))
  150. path_order = {
  151. 1: os.getcwd() + os.sep + file_path_to_test,
  152. 2: "/etc/kalliope" + os.sep + file_path_to_test,
  153. # In this case 'get_current_file_parent_parent_path' is corresponding to kalliope root path
  154. # from /an/unknown/path/kalliope/kalliope/core/Utils to /an/unknown/path/kalliope/kalliope
  155. 3: cls.get_current_file_parent_parent_path(current_script_path) + os.sep + file_path_to_test
  156. }
  157. for key in sorted(path_order):
  158. new_file_path_to_test = path_order[key]
  159. logger.debug("Try to load file from %s: %s" % (key, new_file_path_to_test))
  160. if os.path.isfile(new_file_path_to_test):
  161. logger.debug("File found in %s" % new_file_path_to_test)
  162. return new_file_path_to_test
  163. else:
  164. if os.path.isfile(file_path_to_test):
  165. return file_path_to_test
  166. else:
  167. return None
  168. @staticmethod
  169. def query_yes_no(question, default="yes"):
  170. """Ask a yes/no question via raw_input() and return their answer.
  171. "question" is a string that is presented to the user.
  172. "default" is the presumed answer if the user just hits <Enter>.
  173. It must be "yes" (the default), "no" or None (meaning
  174. an answer is required of the user).
  175. The "answer" return value is True for "yes" or False for "no".
  176. """
  177. valid = {"yes": True, "y": True, "ye": True,
  178. "no": False, "n": False}
  179. if default is None:
  180. prompt = " [y/n] "
  181. elif default == "yes":
  182. prompt = " [Y/n] "
  183. elif default == "no":
  184. prompt = " [y/N] "
  185. else:
  186. raise ValueError("invalid default answer: '%s'" % default)
  187. while True:
  188. Utils.print_warning(question + prompt)
  189. choice = raw_input().lower()
  190. if default is not None and choice == '':
  191. return valid[default]
  192. elif choice in valid:
  193. return valid[choice]
  194. else:
  195. Utils.print_warning("Please respond with 'yes' or 'no' or 'y' or 'n').\n")
  196. ##################
  197. #
  198. # Brackets management
  199. #
  200. #########
  201. @staticmethod
  202. def is_containing_bracket(sentence):
  203. """
  204. Return True if the text in <sentence> contains brackets
  205. :param sentence:
  206. :return:
  207. """
  208. # print "sentence to test %s" % sentence
  209. pattern = r"{{|}}"
  210. # prog = re.compile(pattern)
  211. if not isinstance(sentence, six.text_type):
  212. sentence = str(sentence)
  213. check_bool = re.search(pattern, sentence)
  214. if check_bool is not None:
  215. return True
  216. return False
  217. @staticmethod
  218. def find_all_matching_brackets(sentence):
  219. """
  220. Find all the bracket matches from a given sentence
  221. :param sentence: the sentence to check
  222. :return: the list with all the matches
  223. """
  224. pattern = r"((?:{{\s*)[\w\.]+(?:\s*}}))"
  225. # find everything like {{ word }}
  226. if not isinstance(sentence, six.text_type):
  227. sentence = str(sentence)
  228. return re.findall(pattern, sentence)
  229. @staticmethod
  230. def remove_spaces_in_brackets(sentence):
  231. """
  232. If has brackets it removes spaces in brackets
  233. :param sentence: the sentence to work on
  234. :return: the sentence without any spaces in brackets
  235. """
  236. pattern = '\s+(?=[^\{\{\}\}]*\}\})'
  237. # Remove white spaces (if any) between the variable and the double brace then split
  238. if not isinstance(sentence, six.text_type):
  239. sentence = str(sentence)
  240. return re.sub(pattern, '', sentence)
  241. ##################
  242. #
  243. # Lists management
  244. #
  245. #########
  246. @staticmethod
  247. def get_next_value_list(list_to_check):
  248. ite = list_to_check.__iter__()
  249. next(ite, None)
  250. return next(ite, None)
  251. ##################
  252. #
  253. # Encoding
  254. #
  255. #########
  256. @staticmethod
  257. def encode_text_utf8(text):
  258. if sys.version_info[0] < 3:
  259. if isinstance(text, unicode):
  260. text = text.encode("utf-8")
  261. return text