Run a Twisted reactor inside IPython.
- Copy
twist.pyto~/.ipython/twist.py - Copy
tpythonto~/bin/tpython - Make
tpythonexecutable:chmod +x ~/bin/tpython
You may need to add $HOME/bin to your path.
Last active
July 3, 2018 10:42
-
-
Save kived/8721434 to your computer and use it in GitHub Desktop.
TPython: IPython with a Twisted reactor
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/bin/sh | |
| ipython -i ~/.ipython/twist.py "$@" |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| ''' | |
| IPython Twisted | |
| =============== | |
| Run a Twisted reactor inside IPython. Works with newer IPython versions (tested with 0.13.2). | |
| Logging is automatically enabled via `logging.basicConfig(level=logging.DEBUG)`. | |
| ''' | |
| import logging | |
| logging.basicConfig(level=logging.DEBUG) | |
| def __install(): | |
| log = logging.getLogger('tpython') | |
| log.info('setting up twisted reactor in ipython loop') | |
| from twisted.internet import _threadedselect | |
| _threadedselect.install() | |
| from twisted.internet import reactor | |
| from collections import deque | |
| from IPython.lib import inputhook | |
| from IPython import InteractiveShell | |
| q = deque() | |
| def reactor_wake(twisted_loop_next, q=q): | |
| q.append(twisted_loop_next) | |
| def reactor_work(*_args): | |
| if q: | |
| while len(q): | |
| q.popleft()() | |
| return 0 | |
| def reactor_start(*_args): | |
| log.info('starting twisted reactor in ipython') | |
| reactor.interleave(reactor_wake) # @UndefinedVariable | |
| inputhook.set_inputhook(reactor_work) | |
| def reactor_stop(): | |
| if reactor.threadpool: # @UndefinedVariable | |
| log.info('stopping twisted threads') | |
| reactor.threadpool.stop() # @UndefinedVariable | |
| log.info('shutting down twisted reactor') | |
| reactor._mainLoopShutdown() # @UndefinedVariable | |
| ip = InteractiveShell.instance() | |
| ask_exit = ip.ask_exit | |
| def ipython_exit(): | |
| reactor_stop() | |
| return ask_exit() | |
| ip.ask_exit = ipython_exit | |
| reactor_start() | |
| return reactor | |
| reactor = __install() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment