BrainLoader.py 4.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136
  1. import inspect
  2. import logging
  3. import os
  4. from YAMLLoader import YAMLLoader
  5. from core.ConfigurationManager import Singleton
  6. from core.ConfigurationManager.ConfigurationChecker import ConfigurationChecker
  7. from core.Models.Brain import Brain
  8. from core.Models.Event import Event
  9. from core.Models.Neuron import Neuron
  10. from core.Models.Order import Order
  11. from core.Models.Synapse import Synapse
  12. logging.basicConfig()
  13. logger = logging.getLogger("kalliope")
  14. @Singleton
  15. class BrainLoader(object):
  16. def __init__(self):
  17. # Todo check how to provide the file_path
  18. self.brain = self._get_brain()
  19. self.yaml_config = self._get_yaml_config()
  20. @classmethod
  21. def _get_yaml_config(cls, file_path=None):
  22. if file_path is None:
  23. brain_file_path = cls._get_root_brain_path()
  24. else:
  25. brain_file_path = file_path
  26. return YAMLLoader.get_config(brain_file_path)
  27. @classmethod
  28. def _get_brain(cls, file_path=None):
  29. """
  30. return a brain object from YAML settings
  31. :return: Brain object
  32. :rtype: Brain
  33. """
  34. # get the brain with dict
  35. dict_brain = cls._get_yaml_config(file_path)
  36. # create a new brain
  37. brain = Brain()
  38. brain.brain_yaml = dict_brain
  39. # create list of Synapse
  40. synapses = list()
  41. for synapses_dict in dict_brain:
  42. if "includes" not in synapses_dict: # we don't need to check includes as it's not a synapse
  43. if ConfigurationChecker().check_synape_dict(synapses_dict):
  44. # print "synapses_dict ok"
  45. name = synapses_dict["name"]
  46. neurons = cls._get_neurons(synapses_dict["neurons"])
  47. signals = cls._get_signals(synapses_dict["signals"])
  48. new_synapse = Synapse(name=name, neurons=neurons, signals=signals)
  49. synapses.append(new_synapse)
  50. brain.synapses = synapses
  51. if file_path is None:
  52. brain.brain_file = cls._get_root_brain_path()
  53. else:
  54. brain.brain_file = file_path
  55. # check that no synapse have the same name than another
  56. if ConfigurationChecker().check_synapes(synapses):
  57. return brain
  58. return None
  59. @staticmethod
  60. def _get_neurons(neurons_dict):
  61. """
  62. Get a list of Neuron object from a neuron dict
  63. :param neurons_dict:
  64. :return:
  65. """
  66. neurons = list()
  67. for neuron_dict in neurons_dict:
  68. if isinstance(neuron_dict, dict):
  69. if ConfigurationChecker().check_neuron_dict(neuron_dict):
  70. # print "Neurons dict ok"
  71. for neuron_name in neuron_dict:
  72. name = neuron_name
  73. parameters = neuron_dict[name]
  74. # print parameters
  75. new_neuron = Neuron(name=name, parameters=parameters)
  76. neurons.append(new_neuron)
  77. else:
  78. # the neuron does not have parameter
  79. if ConfigurationChecker().check_neuron_dict(neuron_dict):
  80. new_neuron = Neuron(name=neuron_dict)
  81. neurons.append(new_neuron)
  82. return neurons
  83. @classmethod
  84. def _get_signals(cls, signals_dict):
  85. # print signals_dict
  86. signals = list()
  87. for signal_dict in signals_dict:
  88. if ConfigurationChecker().check_signal_dict(signal_dict):
  89. # print "Signals dict ok"
  90. event_or_order = cls._get_event_or_order_from_dict(signal_dict)
  91. signals.append(event_or_order)
  92. return signals
  93. @staticmethod
  94. def _get_event_or_order_from_dict(signal_or_event_dict):
  95. if 'event' in signal_or_event_dict:
  96. # print "is event"
  97. event = signal_or_event_dict["event"]
  98. if ConfigurationChecker.check_event_dict(event):
  99. return Event(period=event)
  100. if 'order' in signal_or_event_dict:
  101. order = signal_or_event_dict["order"]
  102. if ConfigurationChecker.check_order_dict(order):
  103. return Order(sentence=order)
  104. @staticmethod
  105. def _get_root_brain_path():
  106. """
  107. Return the full path of the default brain file
  108. :return:
  109. """
  110. # get current script directory path. We are in /an/unknown/path/kalliope/core/ConfigurationManager
  111. cur_script_directory = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))
  112. # get parent dir. Now we are in /an/unknown/path/kalliope
  113. parent_dir = os.path.normpath(cur_script_directory + os.sep + os.pardir + os.sep + os.pardir)
  114. brain_path = parent_dir + os.sep + "brain.yml"
  115. logger.debug("Real brain.yml path: %s" % brain_path)
  116. if os.path.isfile(brain_path):
  117. return brain_path
  118. raise IOError("Default brain.yml file not found")