Funktionierender Prototyp des Serious Games zur Vermittlung von Wissen zu Software-Engineering-Arbeitsmodellen.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

wxreactor.py 5.2KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188
  1. # Copyright (c) Twisted Matrix Laboratories.
  2. # See LICENSE for details.
  3. """
  4. This module provides wxPython event loop support for Twisted.
  5. In order to use this support, simply do the following::
  6. | from twisted.internet import wxreactor
  7. | wxreactor.install()
  8. Then, when your root wxApp has been created::
  9. | from twisted.internet import reactor
  10. | reactor.registerWxApp(yourApp)
  11. | reactor.run()
  12. Then use twisted.internet APIs as usual. Stop the event loop using
  13. reactor.stop(), not yourApp.ExitMainLoop().
  14. IMPORTANT: tests will fail when run under this reactor. This is
  15. expected and probably does not reflect on the reactor's ability to run
  16. real applications.
  17. """
  18. from queue import Empty, Queue
  19. try:
  20. from wx import ( # type: ignore[import]
  21. CallAfter as wxCallAfter,
  22. PySimpleApp as wxPySimpleApp,
  23. Timer as wxTimer,
  24. )
  25. except ImportError:
  26. # older version of wxPython:
  27. from wxPython.wx import wxPySimpleApp, wxCallAfter, wxTimer # type: ignore[import]
  28. from twisted.internet import _threadedselect
  29. from twisted.python import log, runtime
  30. class ProcessEventsTimer(wxTimer):
  31. """
  32. Timer that tells wx to process pending events.
  33. This is necessary on macOS, probably due to a bug in wx, if we want
  34. wxCallAfters to be handled when modal dialogs, menus, etc. are open.
  35. """
  36. def __init__(self, wxapp):
  37. wxTimer.__init__(self)
  38. self.wxapp = wxapp
  39. def Notify(self):
  40. """
  41. Called repeatedly by wx event loop.
  42. """
  43. self.wxapp.ProcessPendingEvents()
  44. class WxReactor(_threadedselect.ThreadedSelectReactor):
  45. """
  46. wxPython reactor.
  47. wxPython drives the event loop, select() runs in a thread.
  48. """
  49. _stopping = False
  50. def registerWxApp(self, wxapp):
  51. """
  52. Register wxApp instance with the reactor.
  53. """
  54. self.wxapp = wxapp
  55. def _installSignalHandlersAgain(self):
  56. """
  57. wx sometimes removes our own signal handlers, so re-add them.
  58. """
  59. try:
  60. # make _handleSignals happy:
  61. import signal
  62. signal.signal(signal.SIGINT, signal.default_int_handler)
  63. except ImportError:
  64. return
  65. self._handleSignals()
  66. def stop(self):
  67. """
  68. Stop the reactor.
  69. """
  70. if self._stopping:
  71. return
  72. self._stopping = True
  73. _threadedselect.ThreadedSelectReactor.stop(self)
  74. def _runInMainThread(self, f):
  75. """
  76. Schedule function to run in main wx/Twisted thread.
  77. Called by the select() thread.
  78. """
  79. if hasattr(self, "wxapp"):
  80. wxCallAfter(f)
  81. else:
  82. # wx shutdown but twisted hasn't
  83. self._postQueue.put(f)
  84. def _stopWx(self):
  85. """
  86. Stop the wx event loop if it hasn't already been stopped.
  87. Called during Twisted event loop shutdown.
  88. """
  89. if hasattr(self, "wxapp"):
  90. self.wxapp.ExitMainLoop()
  91. def run(self, installSignalHandlers=True):
  92. """
  93. Start the reactor.
  94. """
  95. self._postQueue = Queue()
  96. if not hasattr(self, "wxapp"):
  97. log.msg(
  98. "registerWxApp() was not called on reactor, "
  99. "registering my own wxApp instance."
  100. )
  101. self.registerWxApp(wxPySimpleApp())
  102. # start select() thread:
  103. self.interleave(
  104. self._runInMainThread, installSignalHandlers=installSignalHandlers
  105. )
  106. if installSignalHandlers:
  107. self.callLater(0, self._installSignalHandlersAgain)
  108. # add cleanup events:
  109. self.addSystemEventTrigger("after", "shutdown", self._stopWx)
  110. self.addSystemEventTrigger(
  111. "after", "shutdown", lambda: self._postQueue.put(None)
  112. )
  113. # On macOS, work around wx bug by starting timer to ensure
  114. # wxCallAfter calls are always processed. We don't wake up as
  115. # often as we could since that uses too much CPU.
  116. if runtime.platform.isMacOSX():
  117. t = ProcessEventsTimer(self.wxapp)
  118. t.Start(2) # wake up every 2ms
  119. self.wxapp.MainLoop()
  120. wxapp = self.wxapp
  121. del self.wxapp
  122. if not self._stopping:
  123. # wx event loop exited without reactor.stop() being
  124. # called. At this point events from select() thread will
  125. # be added to _postQueue, but some may still be waiting
  126. # unprocessed in wx, thus the ProcessPendingEvents()
  127. # below.
  128. self.stop()
  129. wxapp.ProcessPendingEvents() # deal with any queued wxCallAfters
  130. while 1:
  131. try:
  132. f = self._postQueue.get(timeout=0.01)
  133. except Empty:
  134. continue
  135. else:
  136. if f is None:
  137. break
  138. try:
  139. f()
  140. except BaseException:
  141. log.err()
  142. def install():
  143. """
  144. Configure the twisted mainloop to be run inside the wxPython mainloop.
  145. """
  146. reactor = WxReactor()
  147. from twisted.internet.main import installReactor
  148. installReactor(reactor)
  149. return reactor
  150. __all__ = ["install"]