terminals.py 8.4KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246
  1. # coding: utf-8
  2. import collections
  3. from copy import copy
  4. from multiprocessing import Queue
  5. from multiprocessing import Process
  6. from queue import Empty
  7. import time
  8. from synergine2.base import BaseObject
  9. from synergine2.config import Config
  10. from synergine2.log import SynergineLogger
  11. from synergine2.simulation import Subject
  12. from synergine2.simulation import Event
  13. STOP_SIGNAL = '__STOP_SIGNAL__'
  14. class TerminalPackage(BaseObject):
  15. """
  16. TODO: Update this class considering shared data across processes
  17. """
  18. def __init__(
  19. self,
  20. subjects: [Subject]=None,
  21. add_subjects: [Subject]=None,
  22. remove_subjects: [Subject]=None,
  23. events: [Event]=None,
  24. simulation_actions: [tuple]=None,
  25. subject_actions: [tuple]=None,
  26. is_cycle: bool=False,
  27. *args,
  28. **kwargs
  29. ):
  30. self.subjects = subjects
  31. self.add_subjects = add_subjects or []
  32. self.remove_subjects = remove_subjects or []
  33. self.events = events or []
  34. self.simulation_actions = simulation_actions or []
  35. self.subject_actions = subject_actions or []
  36. self.is_cycle = is_cycle
  37. def repr_debug(self) -> str:
  38. subjects = self.subjects or []
  39. return str(dict(
  40. subjects=subjects,
  41. add_subjects=[s.id for s in self.add_subjects],
  42. remove_subjects=[s.id for s in self.remove_subjects],
  43. events=[e.repr_debug() for e in self.events],
  44. simulation_actions=['{}: {}'.format(a.__class__.__name__, p) for a, p in self.simulation_actions],
  45. subject_actions=['{}: {}'.format(a.__class__.__name__, p) for a, p in self.subject_actions],
  46. is_cycle=self.is_cycle,
  47. ))
  48. class Terminal(BaseObject):
  49. # Default behaviour is to do nothing.
  50. # DEFAULT_SLEEP is sleep time between each queue read
  51. default_sleep = 1
  52. # List of subscribed Event classes. Terminal will not receive events
  53. # who are not instance of listed classes
  54. subscribed_events = [Event]
  55. def __init__(
  56. self,
  57. config: Config,
  58. logger: SynergineLogger,
  59. asynchronous: bool=True,
  60. ):
  61. self.config = config
  62. self.logger = logger
  63. self._input_queue = None
  64. self._output_queue = None
  65. self._stop_required = False
  66. self.subjects = {}
  67. self.cycle_events = []
  68. self.event_handlers = collections.defaultdict(list)
  69. self.asynchronous = asynchronous
  70. def accept_event(self, event: Event) -> bool:
  71. for event_class in self.subscribed_events:
  72. if isinstance(event, event_class):
  73. return True
  74. return False
  75. def start(self, input_queue: Queue, output_queue: Queue) -> None:
  76. self._input_queue = input_queue
  77. self._output_queue = output_queue
  78. self.run()
  79. def run(self):
  80. """
  81. Override this method to create your daemon terminal
  82. """
  83. try:
  84. while self.read():
  85. time.sleep(self.default_sleep)
  86. except KeyboardInterrupt:
  87. pass
  88. def read(self):
  89. while True:
  90. try:
  91. package = self._input_queue.get(block=False, timeout=None)
  92. if package == STOP_SIGNAL:
  93. self._stop_required = True
  94. return False
  95. self.receive(package)
  96. except Empty:
  97. return True # Finished to read Queue
  98. def receive(self, package: TerminalPackage):
  99. self.update_with_package(package)
  100. # End of cycle management signal
  101. self.send(TerminalPackage(is_cycle=True))
  102. def send(self, package: TerminalPackage):
  103. self._output_queue.put(package)
  104. def register_event_handler(self, event_class, func):
  105. self.event_handlers[event_class].append(func)
  106. def update_with_package(self, package: TerminalPackage):
  107. if package.subjects:
  108. self.subjects = {s.id: s for s in package.subjects}
  109. for new_subject in package.add_subjects:
  110. self.subjects[new_subject.id] = new_subject
  111. for deleted_subject in package.remove_subjects:
  112. del self.subjects[deleted_subject.id]
  113. self.cycle_events = package.events
  114. self.execute_event_handlers(package.events)
  115. def execute_event_handlers(self, events: [Event]):
  116. for event in events:
  117. for event_class, handlers in self.event_handlers.items():
  118. if isinstance(event, event_class):
  119. for handler in handlers:
  120. handler(event)
  121. class TerminalManager(BaseObject):
  122. def __init__(
  123. self,
  124. config: Config,
  125. logger: SynergineLogger,
  126. terminals: [Terminal]
  127. ):
  128. self.config = config
  129. self.logger = logger
  130. self.terminals = terminals
  131. self.outputs_queues = {}
  132. self.inputs_queues = {}
  133. def start(self) -> None:
  134. self.logger.info('Start terminals')
  135. for terminal in self.terminals:
  136. # TODO: logs
  137. output_queue = Queue()
  138. self.outputs_queues[terminal] = output_queue
  139. input_queue = Queue()
  140. self.inputs_queues[terminal] = input_queue
  141. process = Process(target=terminal.start, kwargs=dict(
  142. input_queue=output_queue,
  143. output_queue=input_queue,
  144. ))
  145. process.start()
  146. def stop(self):
  147. for output_queue in self.outputs_queues.values():
  148. output_queue.put(STOP_SIGNAL)
  149. def send(self, package: TerminalPackage):
  150. self.logger.info('Send package to terminals')
  151. if self.logger.is_debug:
  152. self.logger.debug('Send package to terminals: {}'.format(
  153. str(package.repr_debug()),
  154. ))
  155. for terminal, output_queue in self.outputs_queues.items():
  156. self.logger.info('Send package to terminal {}'.format(terminal.__class__.__name__))
  157. # Terminal maybe don't want all events, so reduce list of event
  158. # Thirst make a copy to personalize this package
  159. terminal_adapted_package = copy(package)
  160. # Duplicate events list to personalize it
  161. terminal_adapted_package.events = terminal_adapted_package.events[:]
  162. for package_event in terminal_adapted_package.events[:]:
  163. if not terminal.accept_event(package_event):
  164. terminal_adapted_package.events.remove(package_event)
  165. if self.logger.is_debug:
  166. self.logger.debug('Send package to terminal {}: {}'.format(
  167. terminal.__class__.__name__,
  168. terminal_adapted_package.repr_debug(),
  169. ))
  170. output_queue.put(terminal_adapted_package)
  171. def receive(self) -> [TerminalPackage]:
  172. self.logger.info('Receive terminals packages')
  173. packages = []
  174. for terminal, input_queue in self.inputs_queues.items():
  175. self.logger.info('Receive terminal {} packages ({})'.format(
  176. terminal.__class__.__name__,
  177. 'sync' if not terminal.asynchronous else 'async'
  178. ))
  179. # When terminal is synchronous, wait it's cycle package
  180. if not terminal.asynchronous:
  181. continue_ = True
  182. while continue_:
  183. package = input_queue.get()
  184. # In case where terminal send package before end of cycle
  185. # management
  186. continue_ = not package.is_cycle
  187. if self.logger.is_debug:
  188. self.logger.debug('Receive package from {}: {}'.format(
  189. terminal.__class__.__name__,
  190. str(package.repr_debug()),
  191. ))
  192. packages.append(package)
  193. else:
  194. try:
  195. while True:
  196. package = input_queue.get(block=False, timeout=None)
  197. if self.logger.is_debug:
  198. self.logger.debug('Receive package from {}: {}'.format(
  199. str(terminal),
  200. str(package.repr_debug()),
  201. ))
  202. packages.append(package)
  203. except Empty:
  204. pass # Queue is empty
  205. self.logger.info('{} package(s) received'.format(len(packages)))
  206. return packages