LIFOBuffer.py 9.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207
  1. import logging
  2. from six import with_metaclass
  3. from kalliope.core.Cortex import Cortex
  4. from kalliope.core.NeuronLauncher import NeuronLauncher
  5. from kalliope.core.Models import Singleton
  6. from kalliope.core.Models.APIResponse import APIResponse
  7. logging.basicConfig()
  8. logger = logging.getLogger("kalliope")
  9. class Serialize(Exception):
  10. """
  11. When raised, the LIFO class return the current API response to the caller
  12. """
  13. pass
  14. class SynapseListAddedToLIFO(Exception):
  15. """
  16. When raised, a synapse list to process has been added to the LIFO list.
  17. The LIFO must start over and process the last synapse list added
  18. """
  19. pass
  20. class LIFOBuffer(with_metaclass(Singleton, object)):
  21. """
  22. This class is a LIFO list of synapse to process where the last synapse list to enter will be the first synapse
  23. list to be processed.
  24. This design is needed in order to use Kalliope from the API.
  25. Because we want to return an information when a Neuron is still processing and waiting for an answer from the user
  26. like with the Neurotransmitter neuron.
  27. """
  28. def __init__(self):
  29. logger.debug("[LIFOBuffer] LIFO buffer created")
  30. self.api_response = APIResponse()
  31. self.lifo_list = list()
  32. self.answer = None
  33. self.is_api_call = False
  34. self.no_voice = False
  35. self.is_running = False
  36. self.reset_lifo = False
  37. def set_answer(self, value):
  38. self.answer = value
  39. def set_api_call(self, value):
  40. self.is_api_call = value
  41. def add_synapse_list_to_lifo(self, matched_synapse_list, high_priority=False):
  42. """
  43. Add a synapse list to process to the lifo
  44. :param matched_synapse_list: List of Matched Synapse
  45. :param high_priority: If True, the synapse list added is executed directly
  46. :return:
  47. """
  48. logger.debug("[LIFOBuffer] Add a new synapse list to process to the LIFO")
  49. self.lifo_list.append(matched_synapse_list)
  50. if high_priority:
  51. self.reset_lifo = True
  52. def clean(self):
  53. """
  54. Clean the LIFO by creating a new list
  55. """
  56. self.lifo_list = list()
  57. self.api_response = APIResponse()
  58. def _return_serialized_api_response(self):
  59. """
  60. Serialize Exception has been raised by the execute process somewhere, return the serialized API response
  61. to the caller. Clean up the APIResponse object for the next call
  62. :return:
  63. """
  64. # we prepare a json response
  65. returned_api_response = self.api_response.serialize()
  66. # we clean up the API response object for the next call
  67. self.api_response = APIResponse()
  68. return returned_api_response
  69. def execute(self, answer=None, is_api_call=False, no_voice=False):
  70. """
  71. Process the LIFO list.
  72. The LIFO list contains multiple list of matched synapses.
  73. For each list of matched synapse we process synapses inside
  74. For each synapses we process neurons.
  75. If a neuron add a Synapse list to the lifo, this synapse list is processed before executing the first list
  76. in which we were in.
  77. :param answer: String answer to give the the last neuron which was waiting for an answer
  78. :param is_api_call: Boolean passed to all neuron in order to let them know if the current call comes from API
  79. :param no_voice: If true, the generated text will not be processed by the TTS engine
  80. :return: serialized APIResponse object
  81. """
  82. # store the answer if present
  83. self.answer = answer
  84. self.is_api_call = is_api_call
  85. self.no_voice = no_voice
  86. if not self.is_running:
  87. self.is_running = True
  88. try:
  89. # we keep looping over the LIFO til we have synapse list to process in it
  90. while self.lifo_list:
  91. logger.debug("[LIFOBuffer] number of synapse list to process: %s" % len(self.lifo_list))
  92. try:
  93. # get the last list of matched synapse in the LIFO
  94. last_synapse_fifo_list = self.lifo_list[-1]
  95. self._process_synapse_list(last_synapse_fifo_list)
  96. except SynapseListAddedToLIFO:
  97. continue
  98. # remove the synapse list from the LIFO
  99. self.lifo_list.remove(last_synapse_fifo_list)
  100. # clean the cortex from value loaded from order as all synapses have been processed
  101. Cortex.clean_parameter_from_order()
  102. self.is_running = False
  103. raise Serialize
  104. except Serialize:
  105. return self._return_serialized_api_response()
  106. def _process_synapse_list(self, synapse_list):
  107. """
  108. Process a list of matched synapse.
  109. Execute each neuron list for each synapse.
  110. Add info in the API response object after each processed synapse
  111. Remove the synapse from the synapse_list when it has been fully executed
  112. :param synapse_list: List of MatchedSynapse
  113. """
  114. # we keep processing til we have synapse in the FIFO to process
  115. while synapse_list:
  116. # get the first matched synapse in the list
  117. matched_synapse = synapse_list[0]
  118. # add the synapse to the API response so the user will get the status if the synapse was not already
  119. # in the response
  120. if matched_synapse not in self.api_response.list_processed_matched_synapse:
  121. self.api_response.list_processed_matched_synapse.append(matched_synapse)
  122. self._process_neuron_list(matched_synapse=matched_synapse)
  123. # The synapse has been processed we can remove it from the list.
  124. synapse_list.remove(matched_synapse)
  125. def _process_neuron_list(self, matched_synapse):
  126. """
  127. Process the neuron list of the matched_synapse
  128. Execute the Neuron
  129. Executing a Neuron creates a NeuronModule object. This one can have 3 status:
  130. - waiting for an answer: The neuron wait for an answer from the caller. The api response object is returned.
  131. The neuron is not removed from the matched synapse to be executed again
  132. - want to execute a synapse: The neuron add a list of synapse to execute to the lifo.
  133. The LIFO restart over to process it.The neuron is removed from the matched synapse
  134. - complete: The neuron has been executed and its not waiting for an answer and doesn't want to start a synapse
  135. The neuron is removed from the matched synapse
  136. :param matched_synapse: MatchedSynapse object to process
  137. """
  138. logger.debug("[LIFOBuffer] number of neuron to process: %s" % len(matched_synapse.neuron_fifo_list))
  139. # while we have synapse to process in the list of synapse
  140. while matched_synapse.neuron_fifo_list:
  141. # get the first neuron in the FIFO neuron list
  142. neuron = matched_synapse.neuron_fifo_list[0]
  143. # from here, we are back into the last neuron we were processing.
  144. if self.answer is not None: # we give the answer if exist to the first neuron
  145. neuron.parameters["answer"] = self.answer
  146. # the next neuron should not get this answer
  147. self.answer = None
  148. # todo fix this when we have a full client/server call. The client would be the voice or api call
  149. neuron.parameters["is_api_call"] = self.is_api_call
  150. neuron.parameters["no_voice"] = self.no_voice
  151. logger.debug("[LIFOBuffer] process_neuron_list: is_api_call: %s, no_voice: %s" % (self.is_api_call,
  152. self.no_voice))
  153. # execute the neuron
  154. instantiated_neuron = NeuronLauncher.start_neuron(neuron=neuron,
  155. parameters_dict=matched_synapse.parameters)
  156. # the status of an execution is "complete" if no neuron are waiting for an answer
  157. self.api_response.status = "complete"
  158. if instantiated_neuron is not None:
  159. if instantiated_neuron.is_waiting_for_answer: # the neuron is waiting for an answer
  160. logger.debug("[LIFOBuffer] Wait for answer mode")
  161. self.api_response.status = "waiting_for_answer"
  162. self.is_running = False
  163. raise Serialize
  164. else:
  165. logger.debug("[LIFOBuffer] complete mode")
  166. # we add the instantiated neuron to the neuron_module_list.
  167. # This one contains info about generated text
  168. matched_synapse.neuron_module_list.append(instantiated_neuron)
  169. # the neuron is fully processed we can remove it from the list
  170. matched_synapse.neuron_fifo_list.remove(neuron)
  171. if self.reset_lifo: # the last executed neuron want to run a synapse
  172. logger.debug("[LIFOBuffer] Last executed neuron want to run a synapse. Restart the LIFO")
  173. # we have added a list of synapse to the LIFO ! this one must start over.
  174. # break all while loop until the execution is back to the LIFO loop
  175. self.reset_lifo = False
  176. raise SynapseListAddedToLIFO
  177. else:
  178. # the neuron has not been processed but we still need to remove it from the list
  179. matched_synapse.neuron_fifo_list.remove(neuron)