123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289 |
- import inspect
- import logging
- import os
- from six import with_metaclass
- from .YAMLLoader import YAMLLoader
- from kalliope.core.Utils import Utils
- from kalliope.core.ConfigurationManager import SettingLoader
- from kalliope.core.ConfigurationManager.ConfigurationChecker import ConfigurationChecker
- from kalliope.core.Models import Singleton
- from kalliope.core.Models.Brain import Brain
- from kalliope.core.Models.Event import Event
- from kalliope.core.Models.Neuron import Neuron
- from kalliope.core.Models.Order import Order
- from kalliope.core.Models.Synapse import Synapse
- logging.basicConfig()
- logger = logging.getLogger("kalliope")
- FILE_NAME = "brain.yml"
- class BrainNotFound(Exception):
- pass
- class BrainLoader(with_metaclass(Singleton, object)):
- """
- This Class is used to get the brain YAML and the Brain as an object
- """
- def __init__(self, file_path=None):
- sl = SettingLoader()
- self.settings = sl.settings
- self.file_path = file_path
- if self.file_path is None: # we don't provide a file path, so search for the default one
- self.file_path = Utils.get_real_file_path(FILE_NAME)
- else:
- self.file_path = Utils.get_real_file_path(file_path)
- # if the returned file path is none, the file doesn't exist
- if self.file_path is None:
- raise BrainNotFound("brain file not found")
- self.yaml_config = self.get_yaml_config()
- self.brain = self.get_brain()
- def get_yaml_config(self):
- """
- Class Methods which loads default or the provided YAML file and return it as a String
- :return: The loaded brain YAML
- :rtype: String
- :Example:
- brain_yaml = BrainLoader.get_yaml_config(/var/tmp/brain.yml)
- .. warnings:: Class Method
- """
- if self.file_path is None:
- brain_file_path = self._get_root_brain_path()
- else:
- brain_file_path = self.file_path
- return YAMLLoader.get_config(brain_file_path)
- def get_brain(self):
- """
- Class Methods which loads default or the provided YAML file and return a Brain
- :return: The loaded Brain
- :rtype: Brain
- :Example:
- brain = BrainLoader.get_brain(file_path="/var/tmp/brain.yml")
- .. seealso:: Brain
- .. warnings:: Class Method
- """
- # Instantiate a brain
- brain = Brain()
- # get the brain with dict
- dict_brain = self.get_yaml_config()
- brain.brain_yaml = dict_brain
- # create list of Synapse
- synapses = list()
- for synapses_dict in dict_brain:
- if "includes" not in synapses_dict: # we don't need to check includes as it's not a synapse
- if ConfigurationChecker().check_synape_dict(synapses_dict):
- name = synapses_dict["name"]
- neurons = self._get_neurons(synapses_dict["neurons"], self.settings)
- signals = self._get_signals(synapses_dict["signals"])
- new_synapse = Synapse(name=name, neurons=neurons, signals=signals)
- synapses.append(new_synapse)
- brain.synapses = synapses
- if self.file_path is None:
- brain.brain_file = self._get_root_brain_path()
- else:
- brain.brain_file = self.file_path
- # check that no synapse have the same name than another
- if not ConfigurationChecker().check_synapes(synapses):
- brain = None
- return brain
- @classmethod
- def _get_neurons(cls, neurons_dict, settings):
- """
- Get a list of Neuron object from a neuron dict
- :param neurons_dict: Neuron name or dictionary of Neuron_name/Neuron_parameters
- :type neurons_dict: String or dict
- :param settings: The Settings with the global variables
- :return: A list of Neurons
- :rtype: List
- :Example:
- neurons = cls._get_neurons(synapses_dict["neurons"])
- .. seealso:: Neuron
- .. warnings:: Static and Private
- """
- neurons = list()
- for neuron_dict in neurons_dict:
- if isinstance(neuron_dict, dict):
- if ConfigurationChecker().check_neuron_dict(neuron_dict):
- for neuron_name in neuron_dict:
- name = neuron_name
- parameters = neuron_dict[name]
- # Update brackets with the global parameter if exist
- parameters = cls._replace_global_variables(parameter=parameters,
- settings=settings)
- new_neuron = Neuron(name=name, parameters=parameters)
- neurons.append(new_neuron)
- else:
- # the neuron does not have parameter
- if ConfigurationChecker().check_neuron_dict(neuron_dict):
- new_neuron = Neuron(name=neuron_dict)
- neurons.append(new_neuron)
- return neurons
- @classmethod
- def _get_signals(cls, signals_dict):
- """
- Get a list of Signal object from a signals dict
- :param signals_dict: Signal name or dictionary of Signal_name/Signal_parameters
- :type signals_dict: String or dict
- :return: A list of Event and/or Order
- :rtype: List
- :Example:
- signals = cls._get_signals(synapses_dict["signals"])
- .. seealso:: Event, Order
- .. warnings:: Class method and Private
- """
- signals = list()
- for signal_dict in signals_dict:
- if ConfigurationChecker().check_signal_dict(signal_dict):
- event_or_order = cls._get_event_or_order_from_dict(signal_dict)
- signals.append(event_or_order)
- return signals
- @classmethod
- def _get_event_or_order_from_dict(cls, signal_or_event_dict):
- """
- The signal is either an Event or an Order
- :param signal_or_event_dict: A dict of event or signal
- :type signal_or_event_dict: dict
- :return: The object corresponding to An Order or an Event
- :rtype: An Order or an Event
- :Example:
- event_or_order = cls._get_event_or_order_from_dict(signal_dict)
- .. seealso:: Event, Order
- .. warnings:: Static method and Private
- """
- if 'event' in signal_or_event_dict:
- event = signal_or_event_dict["event"]
- if ConfigurationChecker.check_event_dict(event):
- return cls._get_event_object(event)
- if 'order' in signal_or_event_dict:
- order = signal_or_event_dict["order"]
- if ConfigurationChecker.check_order_dict(order):
- return Order(sentence=order)
- @staticmethod
- def _get_root_brain_path():
- """
- Return the full path of the default brain file
- :Example:
- brain.brain_file = cls._get_root_brain_path()
- .. raises:: IOError
- .. warnings:: Static method and Private
- """
- # get current script directory path. We are in /an/unknown/path/kalliope/core/ConfigurationManager
- cur_script_directory = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))
- # get parent dir. Now we are in /an/unknown/path/kalliope
- parent_dir = os.path.normpath(cur_script_directory + os.sep + os.pardir + os.sep + os.pardir)
- brain_path = parent_dir + os.sep + "brain.yml"
- logger.debug("Real brain.yml path: %s" % brain_path)
- if os.path.isfile(brain_path):
- return brain_path
- raise IOError("Default brain.yml file not found")
- @classmethod
- def _get_event_object(cls, event_dict):
- def get_key(key_name):
- try:
- return event_dict[key_name]
- except KeyError:
- return None
- year = get_key("year")
- month = get_key("month")
- day = get_key("day")
- week = get_key("week")
- day_of_week = get_key("day_of_week")
- hour = get_key("hour")
- minute = get_key("minute")
- second = get_key("second")
- return Event(year=year, month=month, day=day, week=week,
- day_of_week=day_of_week, hour=hour, minute=minute, second=second)
- @classmethod
- def _replace_global_variables(cls, parameter, settings):
- """
- replace a parameter that contains bracket by the instantiated parameter from the var file
- This function will call itself multiple time to handle different level of parameter in a neuron
- :param parameter: the parameter to update. can be a dict, a list or a string
- :param settings: the settings
- :return: the parameter dict
- """
- if isinstance(parameter, dict):
- for key, value in parameter.items():
- parameter[key] = cls._replace_global_variables(value, settings=settings)
- return parameter
- if isinstance(parameter, list):
- new_parameter_list = list()
- for el in parameter:
- new_parameter_list.append(cls._replace_global_variables(el, settings=settings))
- return new_parameter_list
- if isinstance(parameter, str) or isinstance(parameter, unicode) or isinstance(parameter, int):
- if Utils.is_containing_bracket(parameter):
- return cls._get_global_variable(sentence=parameter, settings=settings)
- return parameter
- @staticmethod
- def _get_global_variable(sentence, settings):
- """
- Get the global variable from the sentence with brackets
- :param sentence: the sentence to check
- :return: the global variable
- """
- sentence_no_spaces = Utils.remove_spaces_in_brackets(sentence=sentence)
- list_of_bracket_params = Utils.find_all_matching_brackets(sentence=sentence_no_spaces)
- for param_with_bracket in list_of_bracket_params:
- param_no_brackets = param_with_bracket.replace("{{", "").replace("}}", "")
- if param_no_brackets in settings.variables:
- logger.debug("Replacing variable %s with %s" % (param_with_bracket,
- settings.variables[param_no_brackets]))
- # need to check if the variable is an integer
- variable = settings.variables[param_no_brackets]
- if isinstance(variable, int):
- variable = str(variable)
- sentence_no_spaces = sentence_no_spaces.replace(param_with_bracket,
- variable)
- return sentence_no_spaces
|