SynapseLauncher.py 4.1 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091
  1. import logging
  2. from kalliope.core.ConfigurationManager import BrainLoader
  3. from kalliope.core.LIFOBuffer import LIFOBuffer
  4. from kalliope.core.Models.MatchedSynapse import MatchedSynapse
  5. from kalliope.core.NeuronLauncher import NeuronLauncher
  6. from kalliope.core.OrderAnalyser import OrderAnalyser
  7. logging.basicConfig()
  8. logger = logging.getLogger("kalliope")
  9. class SynapseNameNotFound(Exception):
  10. """
  11. The Synapse has not been found
  12. .. seealso: Synapse
  13. """
  14. pass
  15. class SynapseLauncher(object):
  16. @classmethod
  17. def start_synapse_by_name(cls, name, brain=None, overriding_parameter_dict=None):
  18. """
  19. Start a synapse by it's name
  20. :param name: Name (Unique ID) of the synapse to launch
  21. :param brain: Brain instance
  22. :param overriding_parameter_dict: parameter to pass to neurons
  23. """
  24. logger.debug("[SynapseLauncher] start_synapse_by_name called with synapse name: %s " % name)
  25. # check if we have found and launched the synapse
  26. synapse = brain.get_synapse_by_name(synapse_name=name)
  27. if not synapse:
  28. raise SynapseNameNotFound("The synapse name \"%s\" does not exist in the brain file" % name)
  29. else:
  30. # get our singleton LIFO
  31. lifo_buffer = LIFOBuffer()
  32. list_synapse_to_process = list()
  33. new_matching_synapse = MatchedSynapse(matched_synapse=synapse,
  34. matched_order=None,
  35. user_order=None,
  36. overriding_parameter=overriding_parameter_dict)
  37. list_synapse_to_process.append(new_matching_synapse)
  38. lifo_buffer.add_synapse_list_to_lifo(list_synapse_to_process)
  39. return lifo_buffer.execute(is_api_call=True)
  40. @classmethod
  41. def run_matching_synapse_from_order(cls, order_to_process, brain, settings, is_api_call=False, no_voice=False):
  42. """
  43. :param order_to_process: the spoken order sent by the user
  44. :param brain: Brain object
  45. :param settings: Settings object
  46. :param is_api_call: if True, the current call come from the API. This info must be known by launched Neuron
  47. :param no_voice: If true, the generated text will not be processed by the TTS engine
  48. :return: list of matched synapse
  49. """
  50. # get our singleton LIFO
  51. lifo_buffer = LIFOBuffer()
  52. # if the LIFO is not empty, so, the current order is passed to the current processing synapse as an answer
  53. if len(lifo_buffer.lifo_list) > 0:
  54. # the LIFO is not empty, this is an answer to a previous call
  55. return lifo_buffer.execute(answer=order_to_process, is_api_call=is_api_call, no_voice=no_voice)
  56. else: # the LIFO is empty, this is a new call
  57. # get a list of matched synapse from the order
  58. list_synapse_to_process = OrderAnalyser.get_matching_synapse(order=order_to_process, brain=brain)
  59. if not list_synapse_to_process: # the order analyser returned us an empty list
  60. # add the default synapse if exist into the lifo
  61. if settings.default_synapse:
  62. logger.debug("[SynapseLauncher] No matching Synapse-> running default synapse ")
  63. # get the default synapse
  64. default_synapse = brain.get_synapse_by_name(settings.default_synapse)
  65. new_matching_synapse = MatchedSynapse(matched_synapse=default_synapse,
  66. matched_order=None,
  67. user_order=order_to_process)
  68. list_synapse_to_process.append(new_matching_synapse)
  69. else:
  70. logger.debug("[SynapseLauncher] No matching Synapse and no default synapse ")
  71. lifo_buffer.add_synapse_list_to_lifo(list_synapse_to_process)
  72. lifo_buffer.api_response.user_order = order_to_process
  73. return lifo_buffer.execute(is_api_call=is_api_call, no_voice=no_voice)