BrainLoader.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289
  1. import inspect
  2. import logging
  3. import os
  4. from six import with_metaclass
  5. from .YAMLLoader import YAMLLoader
  6. from kalliope.core.Utils import Utils
  7. from kalliope.core.ConfigurationManager import SettingLoader
  8. from kalliope.core.ConfigurationManager.ConfigurationChecker import ConfigurationChecker
  9. from kalliope.core.Models import Singleton
  10. from kalliope.core.Models.Brain import Brain
  11. from kalliope.core.Models.Event import Event
  12. from kalliope.core.Models.Neuron import Neuron
  13. from kalliope.core.Models.Order import Order
  14. from kalliope.core.Models.Synapse import Synapse
  15. logging.basicConfig()
  16. logger = logging.getLogger("kalliope")
  17. FILE_NAME = "brain.yml"
  18. class BrainNotFound(Exception):
  19. pass
  20. class BrainLoader(with_metaclass(Singleton, object)):
  21. """
  22. This Class is used to get the brain YAML and the Brain as an object
  23. """
  24. def __init__(self, file_path=None):
  25. sl = SettingLoader()
  26. self.settings = sl.settings
  27. self.file_path = file_path
  28. if self.file_path is None: # we don't provide a file path, so search for the default one
  29. self.file_path = Utils.get_real_file_path(FILE_NAME)
  30. else:
  31. self.file_path = Utils.get_real_file_path(file_path)
  32. # if the returned file path is none, the file doesn't exist
  33. if self.file_path is None:
  34. raise BrainNotFound("brain file not found")
  35. self.yaml_config = self.get_yaml_config()
  36. self.brain = self.get_brain()
  37. def get_yaml_config(self):
  38. """
  39. Class Methods which loads default or the provided YAML file and return it as a String
  40. :return: The loaded brain YAML
  41. :rtype: String
  42. :Example:
  43. brain_yaml = BrainLoader.get_yaml_config(/var/tmp/brain.yml)
  44. .. warnings:: Class Method
  45. """
  46. if self.file_path is None:
  47. brain_file_path = self._get_root_brain_path()
  48. else:
  49. brain_file_path = self.file_path
  50. return YAMLLoader.get_config(brain_file_path)
  51. def get_brain(self):
  52. """
  53. Class Methods which loads default or the provided YAML file and return a Brain
  54. :return: The loaded Brain
  55. :rtype: Brain
  56. :Example:
  57. brain = BrainLoader.get_brain(file_path="/var/tmp/brain.yml")
  58. .. seealso:: Brain
  59. .. warnings:: Class Method
  60. """
  61. # Instantiate a brain
  62. brain = Brain()
  63. # get the brain with dict
  64. dict_brain = self.get_yaml_config()
  65. brain.brain_yaml = dict_brain
  66. # create list of Synapse
  67. synapses = list()
  68. for synapses_dict in dict_brain:
  69. if "includes" not in synapses_dict: # we don't need to check includes as it's not a synapse
  70. if ConfigurationChecker().check_synape_dict(synapses_dict):
  71. name = synapses_dict["name"]
  72. neurons = self._get_neurons(synapses_dict["neurons"], self.settings)
  73. signals = self._get_signals(synapses_dict["signals"])
  74. new_synapse = Synapse(name=name, neurons=neurons, signals=signals)
  75. synapses.append(new_synapse)
  76. brain.synapses = synapses
  77. if self.file_path is None:
  78. brain.brain_file = self._get_root_brain_path()
  79. else:
  80. brain.brain_file = self.file_path
  81. # check that no synapse have the same name than another
  82. if not ConfigurationChecker().check_synapes(synapses):
  83. brain = None
  84. return brain
  85. @classmethod
  86. def _get_neurons(cls, neurons_dict, settings):
  87. """
  88. Get a list of Neuron object from a neuron dict
  89. :param neurons_dict: Neuron name or dictionary of Neuron_name/Neuron_parameters
  90. :type neurons_dict: String or dict
  91. :param settings: The Settings with the global variables
  92. :return: A list of Neurons
  93. :rtype: List
  94. :Example:
  95. neurons = cls._get_neurons(synapses_dict["neurons"])
  96. .. seealso:: Neuron
  97. .. warnings:: Static and Private
  98. """
  99. neurons = list()
  100. for neuron_dict in neurons_dict:
  101. if isinstance(neuron_dict, dict):
  102. if ConfigurationChecker().check_neuron_dict(neuron_dict):
  103. for neuron_name in neuron_dict:
  104. name = neuron_name
  105. parameters = neuron_dict[name]
  106. # Update brackets with the global parameter if exist
  107. parameters = cls._replace_global_variables(parameter=parameters,
  108. settings=settings)
  109. new_neuron = Neuron(name=name, parameters=parameters)
  110. neurons.append(new_neuron)
  111. else:
  112. # the neuron does not have parameter
  113. if ConfigurationChecker().check_neuron_dict(neuron_dict):
  114. new_neuron = Neuron(name=neuron_dict)
  115. neurons.append(new_neuron)
  116. return neurons
  117. @classmethod
  118. def _get_signals(cls, signals_dict):
  119. """
  120. Get a list of Signal object from a signals dict
  121. :param signals_dict: Signal name or dictionary of Signal_name/Signal_parameters
  122. :type signals_dict: String or dict
  123. :return: A list of Event and/or Order
  124. :rtype: List
  125. :Example:
  126. signals = cls._get_signals(synapses_dict["signals"])
  127. .. seealso:: Event, Order
  128. .. warnings:: Class method and Private
  129. """
  130. signals = list()
  131. for signal_dict in signals_dict:
  132. if ConfigurationChecker().check_signal_dict(signal_dict):
  133. event_or_order = cls._get_event_or_order_from_dict(signal_dict)
  134. signals.append(event_or_order)
  135. return signals
  136. @classmethod
  137. def _get_event_or_order_from_dict(cls, signal_or_event_dict):
  138. """
  139. The signal is either an Event or an Order
  140. :param signal_or_event_dict: A dict of event or signal
  141. :type signal_or_event_dict: dict
  142. :return: The object corresponding to An Order or an Event
  143. :rtype: An Order or an Event
  144. :Example:
  145. event_or_order = cls._get_event_or_order_from_dict(signal_dict)
  146. .. seealso:: Event, Order
  147. .. warnings:: Static method and Private
  148. """
  149. if 'event' in signal_or_event_dict:
  150. event = signal_or_event_dict["event"]
  151. if ConfigurationChecker.check_event_dict(event):
  152. return cls._get_event_object(event)
  153. if 'order' in signal_or_event_dict:
  154. order = signal_or_event_dict["order"]
  155. if ConfigurationChecker.check_order_dict(order):
  156. return Order(sentence=order)
  157. @staticmethod
  158. def _get_root_brain_path():
  159. """
  160. Return the full path of the default brain file
  161. :Example:
  162. brain.brain_file = cls._get_root_brain_path()
  163. .. raises:: IOError
  164. .. warnings:: Static method and Private
  165. """
  166. # get current script directory path. We are in /an/unknown/path/kalliope/core/ConfigurationManager
  167. cur_script_directory = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))
  168. # get parent dir. Now we are in /an/unknown/path/kalliope
  169. parent_dir = os.path.normpath(cur_script_directory + os.sep + os.pardir + os.sep + os.pardir)
  170. brain_path = parent_dir + os.sep + "brain.yml"
  171. logger.debug("Real brain.yml path: %s" % brain_path)
  172. if os.path.isfile(brain_path):
  173. return brain_path
  174. raise IOError("Default brain.yml file not found")
  175. @classmethod
  176. def _get_event_object(cls, event_dict):
  177. def get_key(key_name):
  178. try:
  179. return event_dict[key_name]
  180. except KeyError:
  181. return None
  182. year = get_key("year")
  183. month = get_key("month")
  184. day = get_key("day")
  185. week = get_key("week")
  186. day_of_week = get_key("day_of_week")
  187. hour = get_key("hour")
  188. minute = get_key("minute")
  189. second = get_key("second")
  190. return Event(year=year, month=month, day=day, week=week,
  191. day_of_week=day_of_week, hour=hour, minute=minute, second=second)
  192. @classmethod
  193. def _replace_global_variables(cls, parameter, settings):
  194. """
  195. replace a parameter that contains bracket by the instantiated parameter from the var file
  196. This function will call itself multiple time to handle different level of parameter in a neuron
  197. :param parameter: the parameter to update. can be a dict, a list or a string
  198. :param settings: the settings
  199. :return: the parameter dict
  200. """
  201. if isinstance(parameter, dict):
  202. for key, value in parameter.items():
  203. parameter[key] = cls._replace_global_variables(value, settings=settings)
  204. return parameter
  205. if isinstance(parameter, list):
  206. new_parameter_list = list()
  207. for el in parameter:
  208. new_parameter_list.append(cls._replace_global_variables(el, settings=settings))
  209. return new_parameter_list
  210. if isinstance(parameter, str) or isinstance(parameter, unicode) or isinstance(parameter, int):
  211. if Utils.is_containing_bracket(parameter):
  212. return cls._get_global_variable(sentence=parameter, settings=settings)
  213. return parameter
  214. @staticmethod
  215. def _get_global_variable(sentence, settings):
  216. """
  217. Get the global variable from the sentence with brackets
  218. :param sentence: the sentence to check
  219. :return: the global variable
  220. """
  221. sentence_no_spaces = Utils.remove_spaces_in_brackets(sentence=sentence)
  222. list_of_bracket_params = Utils.find_all_matching_brackets(sentence=sentence_no_spaces)
  223. for param_with_bracket in list_of_bracket_params:
  224. param_no_brackets = param_with_bracket.replace("{{", "").replace("}}", "")
  225. if param_no_brackets in settings.variables:
  226. logger.debug("Replacing variable %s with %s" % (param_with_bracket,
  227. settings.variables[param_no_brackets]))
  228. # need to check if the variable is an integer
  229. variable = settings.variables[param_no_brackets]
  230. if isinstance(variable, int):
  231. variable = str(variable)
  232. sentence_no_spaces = sentence_no_spaces.replace(param_with_bracket,
  233. variable)
  234. return sentence_no_spaces