BrainLoader.py 11 KB

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