LIFOBuffer.py 9.4 KB

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