1  import asyncio
   2  import os
   3  import sys
   4  
   5  from IPython.core.debugger import Pdb
   6  from IPython.core.completer import IPCompleter
   7  from .ptutils import IPythonPTCompleter
   8  from .shortcuts import create_ipython_shortcuts
   9  from . import embed
  10  
  11  from pathlib import Path
  12  from pygments.token import Token
  13  from prompt_toolkit.application import create_app_session
  14  from prompt_toolkit.shortcuts.prompt import PromptSession
  15  from prompt_toolkit.enums import EditingMode
  16  from prompt_toolkit.formatted_text import PygmentsTokens
  17  from prompt_toolkit.history import InMemoryHistory, FileHistory
  18  from concurrent.futures import ThreadPoolExecutor
  19  
  20  from prompt_toolkit import __version__ as ptk_version
  21  PTK3 = ptk_version.startswith('3.')
  22  
  23  
  24  # we want to avoid ptk as much as possible when using subprocesses
  25  # as it uses cursor positioning requests, deletes color ....
  26  _use_simple_prompt = "IPY_TEST_SIMPLE_PROMPT" in os.environ
  27  
  28  
  29  class TerminalPdb(Pdb):
  30      """Standalone IPython debugger."""
  31  
  32      def __init__(self, *args, pt_session_options=None, **kwargs):
  33          Pdb.__init__(self, *args, **kwargs)
  34          self._ptcomp = None
  35          self.pt_init(pt_session_options)
  36          self.thread_executor = ThreadPoolExecutor(1)
  37  
  38      def pt_init(self, pt_session_options=None):
  39          """Initialize the prompt session and the prompt loop
  40          and store them in self.pt_app and self.pt_loop.
  41  
  42          Additional keyword arguments for the PromptSession class
  43          can be specified in pt_session_options.
  44          """
  45          if pt_session_options is None:
  46              pt_session_options = {}
  47  
  48          def get_prompt_tokens():
  49              return [(Token.Prompt, self.prompt)]
  50  
  51          if self._ptcomp is None:
  52              compl = IPCompleter(
  53                  shell=self.shell, namespace={}, global_namespace={}, parent=self.shell
  54              )
  55              # add a completer for all the do_ methods
  56              methods_names = [m[3:] for m in dir(self) if m.startswith("do_")]
  57  
  58              def gen_comp(self, text):
  59                  return [m for m in methods_names if m.startswith(text)]
  60              import types
  61              newcomp = types.MethodType(gen_comp, compl)
  62              compl.custom_matchers.insert(0, newcomp)
  63              # end add completer.
  64  
  65              self._ptcomp = IPythonPTCompleter(compl)
  66  
  67          # setup history only when we start pdb
  68          if self.shell.debugger_history is None:
  69              if self.shell.debugger_history_file is not None:
  70                  p = Path(self.shell.debugger_history_file).expanduser()
  71                  if not p.exists():
  72                      p.touch()
  73                  self.debugger_history = FileHistory(os.path.expanduser(str(p)))
  74              else:
  75                  self.debugger_history = InMemoryHistory()
  76          else:
  77              self.debugger_history = self.shell.debugger_history
  78  
  79          options = dict(
  80              message=(lambda: PygmentsTokens(get_prompt_tokens())),
  81              editing_mode=getattr(EditingMode, self.shell.editing_mode.upper()),
  82              key_bindings=create_ipython_shortcuts(self.shell),
  83              history=self.debugger_history,
  84              completer=self._ptcomp,
  85              enable_history_search=True,
  86              mouse_support=self.shell.mouse_support,
  87              complete_style=self.shell.pt_complete_style,
  88              style=getattr(self.shell, "style", None),
  89              color_depth=self.shell.color_depth,
  90          )
  91  
  92          if not PTK3:
  93              options['inputhook'] = self.shell.inputhook
  94          options.update(pt_session_options)
  95          if not _use_simple_prompt:
  96              self.pt_loop = asyncio.new_event_loop()
  97              self.pt_app = PromptSession(**options)
  98  
  99      def _prompt(self):
 100          """
 101          In case other prompt_toolkit apps have to run in parallel to this one (e.g. in madbg),
 102          create_app_session must be used to prevent mixing up between them. According to the prompt_toolkit docs:
 103  
 104          > If you need multiple applications running at the same time, you have to create a separate
 105          > `AppSession` using a `with create_app_session():` block.
 106          """
 107          with create_app_session():
 108              return self.pt_app.prompt()
 109  
 110      def cmdloop(self, intro=None):
 111          """Repeatedly issue a prompt, accept input, parse an initial prefix
 112          off the received input, and dispatch to action methods, passing them
 113          the remainder of the line as argument.
 114  
 115          override the same methods from cmd.Cmd to provide prompt toolkit replacement.
 116          """
 117          if not self.use_rawinput:
 118              raise ValueError('Sorry ipdb does not support use_rawinput=False')
 119  
 120          # In order to make sure that prompt, which uses asyncio doesn't
 121          # interfere with applications in which it's used, we always run the
 122          # prompt itself in a different thread (we can't start an event loop
 123          # within an event loop). This new thread won't have any event loop
 124          # running, and here we run our prompt-loop.
 125          self.preloop()
 126  
 127          try:
 128              if intro is not None:
 129                  self.intro = intro
 130              if self.intro:
 131                  print(self.intro, file=self.stdout)
 132              stop = None
 133              while not stop:
 134                  if self.cmdqueue:
 135                      line = self.cmdqueue.pop(0)
 136                  else:
 137                      self._ptcomp.ipy_completer.namespace = self.curframe_locals
 138                      self._ptcomp.ipy_completer.global_namespace = self.curframe.f_globals
 139  
 140                      # Run the prompt in a different thread.
 141                      if not _use_simple_prompt:
 142                          try:
 143                              line = self.thread_executor.submit(self._prompt).result()
 144                          except EOFError:
 145                              line = "EOF"
 146                      else:
 147                          line = input("ipdb> ")
 148  
 149                  line = self.precmd(line)
 150                  stop = self.onecmd(line)
 151                  stop = self.postcmd(stop, line)
 152              self.postloop()
 153          except Exception:
 154              raise
 155  
 156      def do_interact(self, arg):
 157          ipshell = embed.InteractiveShellEmbed(
 158              config=self.shell.config,
 159              banner1="*interactive*",
 160              exit_msg="*exiting interactive console...*",
 161          )
 162          global_ns = self.curframe.f_globals
 163          ipshell(
 164              module=sys.modules.get(global_ns["__name__"], None),
 165              local_ns=self.curframe_locals,
 166          )
 167  
 168  
 169  def set_trace(frame=None):
 170      """
 171      Start debugging from `frame`.
 172  
 173      If frame is not specified, debugging starts from caller's frame.
 174      """
 175      TerminalPdb().set_trace(frame or sys._getframe().f_back)
 176  
 177  
 178  if __name__ == '__main__':
 179      import pdb
 180      # IPython.core.debugger.Pdb.trace_dispatch shall not catch
 181      # bdb.BdbQuit. When started through __main__ and an exception
 182      # happened after hitting "c", this is needed in order to
 183      # be able to quit the debugging session (see #9950).
 184      old_trace_dispatch = pdb.Pdb.trace_dispatch
 185      pdb.Pdb = TerminalPdb  # type: ignore
 186      pdb.Pdb.trace_dispatch = old_trace_dispatch  # type: ignore
 187      pdb.main()
 188