Python code 模块，InteractiveConsole() 实例源码

def _run(self):
        try:
            try:
                console = InteractiveConsole(self.locals)
                # __builtins__ may either be the __builtin__ module or
                # __builtin__.__dict__ in the latter case typing
                # locals() at the backdoor prompt spews out lots of
                # useless stuff
                import __builtin__
                console.locals["__builtins__"] = __builtin__
                console.interact(banner=self.banner)
            except SystemExit:  # raised by quit()
                sys.exc_clear()
        finally:
            self.switch_out()
            self.finalize()

def _run(self):
        try:
            try:
                console = InteractiveConsole(self.locals)
                # __builtins__ may either be the __builtin__ module or
                # __builtin__.__dict__ in the latter case typing
                # locals() at the backdoor prompt spews out lots of
                # useless stuff
                import __builtin__
                console.locals["__builtins__"] = __builtin__
                console.interact(banner=self.banner)
            except SystemExit:  # raised by quit()
                sys.exc_clear()
        finally:
            self.switch_out()
            self.finalize()

def _run(self):
        try:
            try:
                console = InteractiveConsole(self.locals)
                # __builtins__ may either be the __builtin__ module or
                # __builtin__.__dict__ in the latter case typing
                # locals() at the backdoor prompt spews out lots of
                # useless stuff
                import __builtin__
                console.locals["__builtins__"] = __builtin__
                console.interact(banner=self.banner)
            except SystemExit:  # raised by quit()
                sys.exc_clear()
        finally:
            self.switch_out()
            self.finalize()

def _run(self):
        try:
            try:
                console = InteractiveConsole(self.locals)
                # __builtins__ may either be the __builtin__ module or
                # __builtin__.__dict__ in the latter case typing
                # locals() at the backdoor prompt spews out lots of
                # useless stuff
                import __builtin__
                console.locals["__builtins__"] = __builtin__
                console.interact(banner=self.banner)
            except SystemExit:  # raised by quit()
                sys.exc_clear()
        finally:
            self.switch_out()
            self.finalize()

def _create_interactive_locals(self):
        # Create and return a *new* locals dictionary based on self.locals,
        # and set any new entries in it. (InteractiveConsole does not
        # copy its locals value)
        _locals = self.locals.copy()
        # __builtins__ may either be the __builtin__ module or
        # __builtin__.__dict__; in the latter case typing
        # locals() at the backdoor prompt spews out lots of
        # useless stuff
        try:
            import __builtin__
            _locals["__builtins__"] = __builtin__
        except ImportError:
            import builtins
            _locals["builtins"] = builtins
            _locals['__builtins__'] = builtins
        return _locals

def handle(self, conn, address):
        """
        Interact with one remote user.

        .. versionchanged:: 1.1b2 Each connection gets its own
            ``locals`` dictionary. Previously they were shared in a
            potentially unsafe manner.
        """
        fobj = conn.makefile(mode="rw")
        fobj = _fileobject(conn, fobj, self.stderr)
        getcurrent()._fileobj = fobj

        getcurrent().switch_in()
        try:
            console = InteractiveConsole(self._create_interactive_locals())
            console.interact(banner=self.banner)
        except SystemExit:  # raised by quit()
            if hasattr(sys, 'exc_clear'): # py2
                sys.exc_clear()
        finally:
            conn.close()
            fobj.close()

def _create_interactive_locals(self):
        # Create and return a *new* locals dictionary based on self.locals,
        # and set any new entries in it. (InteractiveConsole does not
        # copy its locals value)
        _locals = self.locals.copy()
        # __builtins__ may either be the __builtin__ module or
        # __builtin__.__dict__; in the latter case typing
        # locals() at the backdoor prompt spews out lots of
        # useless stuff
        try:
            import __builtin__
            _locals["__builtins__"] = __builtin__
        except ImportError:
            import builtins # pylint:disable=import-error
            _locals["builtins"] = builtins
            _locals['__builtins__'] = builtins
        return _locals

def startSession_(self, startupCode):
        namespace = dict(namespaceInjections)
        if startupCode is not None:
            try:
                code = compile(startupCode, "", "exec", 0)
            except:
                traceback.print_exc(0)
            else:
                try:
                    exec code in namespace
                except:
                    etype, value, tb = sys.exc_info()
                    if tb.tb_next is not None:
                        tb = tb.tb_next
                    traceback.print_exception(etype, value, tb)
                    etype = value = tb = None
        self._console = InteractiveConsole(locals=namespace)

def main():
    rospy.init_node("evaluation_ac")
    ac = ACControllerSimulator()
    rospy.Subscriber("/aide/ac_control", Bool, callback=lambda msg: ac.increase_ac() if msg.data else ac.decrease_ac())

    console = Console()
    console.preprocess = lambda source: source[3:]
    atexit.register(loginfo, "Going down by user-input.")
    ac_thread = Thread(target=ac.simulate)
    ac_thread.daemon = True
    pub_thread = Thread(target=publish, args=(ac, ))
    pub_thread.daemon = True
    pub_thread.start()
    while not rospy.is_shutdown():
        try:
            command = console.raw_input("ac> ")
            if command == "start":
                ac_thread.start()
            if command == "end":
                return

        except EOFError:
            print("")
            ac.finished = True
            rospy.signal_shutdown("Going down.")

def my_debugger(vars):
    '''
    Starts interactive Python terminal at location in script
    call with
    my_debugger(globals().copy())
    anywhere in your script
    or call
    my_debugger(locals().copy())
    from anywhere within this package
    '''
    import readline # optional, will allow Up/Down/History in the console
    import code
    # vars = globals().copy() # in python "global" variables are actually module-level
    vars.update(locals())
    shell = code.InteractiveConsole(vars)
    shell.interact()

def my_debugger(vars):
    '''
    starts interactive Python terminal at location in script
    very handy for debugging
    call this function with
    my_debugger(globals().copy())
    anywhere in the body of the script, or
    my_debugger(locals().copy())
    within a script function
    '''
    import readline # optional, will allow Up/Down/History in the console
    import code
    # vars = globals().copy() # in python "global" variables are actually module-level
    vars.update(locals())
    shell = code.InteractiveConsole(vars)
    shell.interact()

def setup_thread_debugger():
    """Setup a thread debbuger for pytest session

    This function, based on http://stackoverflow.com/a/133384, is meant to
    add debugging facility to pytest that will allow to debug deadlock that
    may sometimes occur.
    """
    def debug(signal, frame):
        """Interrupt running process and provide a python prompt for
        interactive debugging."""
        d = {'_frame': frame}  # Allow access to frame object.
        d.update(frame.f_globals)  # Unless shadowed by global
        d.update(frame.f_locals)

        i = code.InteractiveConsole(d)
        message = "Signal received : entering python shell.\nTraceback:\n"
        message += ''.join(traceback.format_stack(frame))
        i.interact(message)

    signal.signal(signal.SIGUSR1, debug)  # Register handler

def _run(self):
        try:
            try:
                console = InteractiveConsole(self.locals)
                # __builtins__ may either be the __builtin__ module or
                # __builtin__.__dict__ in the latter case typing
                # locals() at the backdoor prompt spews out lots of
                # useless stuff
                import __builtin__
                console.locals["__builtins__"] = __builtin__
                console.interact(banner=self.banner)
            except SystemExit:  # raised by quit()
                sys.exc_clear()
        finally:
            self.switch_out()
            self.finalize()

def _run(self):
        try:
            try:
                console = InteractiveConsole(self.locals)
                # __builtins__ may either be the __builtin__ module or
                # __builtin__.__dict__ in the latter case typing
                # locals() at the backdoor prompt spews out lots of
                # useless stuff
                import __builtin__
                console.locals["__builtins__"] = __builtin__
                console.interact(banner=self.banner)
            except SystemExit:  # raised by quit()
                sys.exc_clear()
        finally:
            self.switch_out()
            self.finalize()

def invoke(cls, ns, banner):  # pragma: nocover
        """
        :param ns: local namespace
        :param banner: interactive shell startup banner

        Embed an interactive native python shell.
        """
        import code
        py_prefix = sys.platform.startswith('java') and 'J' or 'P'
        shell_banner = 'Pecan Interactive Shell\n%sython %s\n\n' % \
            (py_prefix, sys.version)
        shell = code.InteractiveConsole(locals=ns)
        try:
            import readline  # noqa
        except ImportError:
            pass
        shell.interact(shell_banner + banner)

def _run(self):
        try:
            try:
                console = InteractiveConsole(self.locals)
                # __builtins__ may either be the __builtin__ module or
                # __builtin__.__dict__ in the latter case typing
                # locals() at the backdoor prompt spews out lots of
                # useless stuff
                import __builtin__
                console.locals["__builtins__"] = __builtin__
                console.interact(banner=self.banner)
            except SystemExit:  # raised by quit()
                sys.exc_clear()
        finally:
            self.switch_out()
            self.finalize()

def _create_interactive_locals(self):
        # Create and return a *new* locals dictionary based on self.locals,
        # and set any new entries in it. (InteractiveConsole does not
        # copy its locals value)
        _locals = self.locals.copy()
        # __builtins__ may either be the __builtin__ module or
        # __builtin__.__dict__; in the latter case typing
        # locals() at the backdoor prompt spews out lots of
        # useless stuff
        try:
            import __builtin__
            _locals["__builtins__"] = __builtin__
        except ImportError:
            import builtins # pylint:disable=import-error
            _locals["builtins"] = builtins
            _locals['__builtins__'] = builtins
        return _locals

def _run(self):
        try:
            try:
                console = InteractiveConsole(self.locals)
                # __builtins__ may either be the __builtin__ module or
                # __builtin__.__dict__ in the latter case typing
                # locals() at the backdoor prompt spews out lots of
                # useless stuff
                import __builtin__
                console.locals["__builtins__"] = __builtin__
                console.interact(banner=self.banner)
            except SystemExit:  # raised by quit()
                sys.exc_clear()
        finally:
            self.switch_out()
            self.finalize()

def _run(self):
        try:
            try:
                console = InteractiveConsole(self.locals)
                # __builtins__ may either be the __builtin__ module or
                # __builtin__.__dict__ in the latter case typing
                # locals() at the backdoor prompt spews out lots of
                # useless stuff
                import __builtin__
                console.locals["__builtins__"] = __builtin__
                console.interact(banner=self.banner)
            except SystemExit:  # raised by quit()
                sys.exc_clear()
        finally:
            self.switch_out()
            self.finalize()

def _create_interactive_locals(self):
        # Create and return a *new* locals dictionary based on self.locals,
        # and set any new entries in it. (InteractiveConsole does not
        # copy its locals value)
        _locals = self.locals.copy()
        # __builtins__ may either be the __builtin__ module or
        # __builtin__.__dict__; in the latter case typing
        # locals() at the backdoor prompt spews out lots of
        # useless stuff
        try:
            import __builtin__
            _locals["__builtins__"] = __builtin__
        except ImportError:
            import builtins
            _locals["builtins"] = builtins
            _locals['__builtins__'] = builtins
        return _locals

def handle(self, conn, address):
        """
        Interact with one remote user.

        .. versionchanged:: 1.1b2 Each connection gets its own
            ``locals`` dictionary. Previously they were shared in a
            potentially unsafe manner.
        """
        fobj = conn.makefile(mode="rw")
        fobj = _fileobject(conn, fobj, self.stderr)
        getcurrent()._fileobj = fobj

        getcurrent().switch_in()
        try:
            console = InteractiveConsole(self._create_interactive_locals())
            console.interact(banner=self.banner)
        except SystemExit:  # raised by quit()
            if hasattr(sys, 'exc_clear'): # py2
                sys.exc_clear()
        finally:
            conn.close()
            fobj.close()

def push(self, data):
        lines = data.split('\n')
        if self._echo:
            for line in lines:
                self._echo("%s %s\n" % (self.prompt, line))
            c = Chronometer()

        #print 'execute', data    
        # Capture stdout/stderr output as well as code interaction.
        stdout, stderr = sys.stdout, sys.stderr
        sys.stdout = sys.stderr = self
        for line in lines:
            more = code.InteractiveConsole.push(self, line)
            self.prompt = "..." if more else ">>>"

        if self._echo:
            self._echo("%s \n\n" % (c.end()))

        sys.stdout, sys.stderr = stdout, stderr

def __debugTap(sig, frame):
    import code, traceback

    """Interrupt running process, and provide a python prompt for
    interactive debugging."""
    d={'_frame':frame}         # Allow access to frame object.
    d.update(frame.f_globals)  # Unless shadowed by global
    d.update(frame.f_locals)

    i = code.InteractiveConsole(d)
    message  = "Signal received : entering python shell.\nTraceback:\n"
    message += ''.join(traceback.format_stack(frame))
    i.interact(message)

def serv_command_get(**kwargs):
    print("Listing services")
    servs = find_services_from_kwargs(**kwargs)
    code.InteractiveConsole(locals()).interact()

def __init__(self, controller):
    self._received_events = []

    code.InteractiveConsole.__init__(self, {
      'stem': stem,
      'stem.control': stem.control,
      'controller': controller,
      'events': self.get_events,
    })

    self._controller = controller
    self._run_python_commands = True

    # Indicates if we're processing a multiline command, such as conditional
    # block or loop.

    self.is_multiline_context = False

    # Intercept events our controller hears about at a pretty low level since
    # the user will likely be requesting them by direct 'SETEVENTS' calls.

    handle_event_real = self._controller._handle_event

    def handle_event_wrapper(event_message):
      handle_event_real(event_message)
      self._received_events.append(event_message)

    self._controller._handle_event = handle_event_wrapper

def run_multiline_interactive_console(mainmodule=None):
    import code
    import __main__
    mainmodule = mainmodule or __main__
    console = code.InteractiveConsole(mainmodule.__dict__, filename='<stdin>')

    def more_lines(unicodetext):
        # ooh, look at the hack:
        src = "#coding:utf-8\n"+unicodetext.encode('utf-8')
        try:
            code = console.compile(src, '<stdin>', 'single')
        except (OverflowError, SyntaxError, ValueError):
            return False
        else:
            return code is None

    while 1:
        try:
            ps1 = getattr(sys, 'ps1', '>>> ')
            ps2 = getattr(sys, 'ps2', '... ')
            try:
                statement = multiline_input(more_lines, ps1, ps2,
                                            returns_unicode=True)
            except EOFError:
                break
            more = console.push(statement)
            assert not more
        except KeyboardInterrupt:
            console.write("\nKeyboardInterrupt\n")
            console.resetbuffer()

def __init__(self):
        local_ns={'exit':new_exit}
        self.console=code.InteractiveConsole(local_ns)
        self.completer=PythonCompleter.PythonCompleter(global_ns=globals(), local_ns=local_ns).complete

def __init__(self, controller):
    self._received_events = []

    code.InteractiveConsole.__init__(self, {
      'stem': stem,
      'stem.control': stem.control,
      'controller': controller,
      'events': self.get_events,
    })

    self._controller = controller
    self._run_python_commands = True

    # Indicates if we're processing a multiline command, such as conditional
    # block or loop.

    self.is_multiline_context = False

    # Intercept events our controller hears about at a pretty low level since
    # the user will likely be requesting them by direct 'SETEVENTS' calls.

    handle_event_real = self._controller._handle_event

    def handle_event_wrapper(event_message):
      handle_event_real(event_message)
      self._received_events.append(event_message)

    self._controller._handle_event = handle_event_wrapper

def my_debugger():
    # starts interactive Python terminal at location in script
    # call with my_debugger() anywhere in your script
    import readline # optional, will allow Up/Down/History in the console
    import code
    vars = globals().copy()
    vars.update(locals())
    shell = code.InteractiveConsole(vars)
    shell.interact()

def my_debugger():
    # starts interactive Python terminal at location in script
    # call with my_debugger() anywhere in your script
    import readline # optional, will allow Up/Down/History in the console
    import code
    vars = globals().copy()
    vars.update(locals())
    shell = code.InteractiveConsole(vars)
    shell.interact()

def my_debugger(vars):
    # starts interactive Python terminal at location in script
    # call with pl.my_debugger(globals().copy()) anywhere in your script
    # or call my_debugger(locals().copy()) from anywhere within this package or another function
    import readline # optional, will allow Up/Down/History in the console
    import code
    # vars = globals().copy() # in python "global" variables are actually module-level
    vars.update(locals())
    shell = code.InteractiveConsole(vars)
    shell.interact()

def my_debugger(vars):
    # starts interactive Python terminal at location in script
    # call with pl.my_debugger(globals().copy()) anywhere in your script
    # or call my_debugger(locals().copy()) from anywhere within this package or another function
    import readline # optional, will allow Up/Down/History in the console
    import code
    # vars = globals().copy() # in python "global" variables are actually module-level
    vars.update(locals())
    shell = code.InteractiveConsole(vars)
    shell.interact()

def my_debugger():
    # starts interactive Python terminal at location in script
    # call with my_debugger() anywhere in your script
    import readline # optional, will allow Up/Down/History in the console
    import code
    vars = globals().copy()
    vars.update(locals())
    shell = code.InteractiveConsole(vars)
    shell.interact()

def my_debugger():
    # starts interactive Python terminal at location in script
    # call with my_debugger() anywhere in your script
    import readline # optional, will allow Up/Down/History in the console
    import code
    vars = globals().copy()
    vars.update(locals())
    shell = code.InteractiveConsole(vars)
    shell.interact()

def my_debugger(vars):
    # starts interactive Python terminal at location in script
    # call with pl.my_debugger(globals().copy()) anywhere in your script
    # or call my_debugger(locals().copy()) from anywhere within this package or another function
    import readline # optional, will allow Up/Down/History in the console
    import code
    # vars = globals().copy() # in python "global" variables are actually module-level
    vars.update(locals())
    shell = code.InteractiveConsole(vars)
    shell.interact()

def my_debugger(vars):
    # starts interactive Python terminal at location in script
    # call with pl.my_debugger(globals().copy()) anywhere in your script
    # or call my_debugger(locals().copy()) from anywhere within this package or another function
    import readline # optional, will allow Up/Down/History in the console
    import code
    # vars = globals().copy() # in python "global" variables are actually module-level
    vars.update(locals())
    shell = code.InteractiveConsole(vars)
    shell.interact()

def my_debugger():
    # starts interactive Python terminal at location in script
    # call with my_debugger() anywhere in your script
    import readline # optional, will allow Up/Down/History in the console
    import code
    vars = globals().copy()
    vars.update(locals())
    shell = code.InteractiveConsole(vars)
    shell.interact()

def my_debugger(vars):
    # starts interactive Python terminal at location in script
    # call with pl.my_debugger(globals().copy()) anywhere in your script
    # or call my_debugger(locals().copy()) from anywhere within this package or another function
    import readline # optional, will allow Up/Down/History in the console
    import code
    # vars = globals().copy() # in python "global" variables are actually module-level
    vars.update(locals())
    shell = code.InteractiveConsole(vars)
    shell.interact()

def _run(self):
        try:
            try:
                console = InteractiveConsole(self.locals)
                console.interact()
            except SystemExit:  # raised by quit()
                sys.exc_clear()
        finally:
            self.switch_out()
            self.finalize()

def _run(self):
        try:
            try:
                console = InteractiveConsole(self.locals)
                console.interact()
            except SystemExit:  # raised by quit()
                sys.exc_clear()
        finally:
            self.switch_out()
            self.finalize()

def __init__(self):
        local_ns={'exit':new_exit}
        self.console=code.InteractiveConsole(local_ns)
        self.completer=PythonCompleter.PythonCompleter(global_ns=globals(), local_ns=local_ns).complete

def __init__(self, locals=None, filename="<console>",
                 histfile=os.path.expanduser("~/.console-history")):
        code.InteractiveConsole.__init__(self, locals, filename)
        self.init_history(histfile)

def __init__(self, locals=None, filename="<console>", histfile=os.path.expanduser("~/.console-history")):
        code.InteractiveConsole.__init__(self, locals, filename)
        self.init_history(histfile)

def debug(sig, frame):
    """Interrupt running process, and provide a python prompt for
    interactive debugging."""
    d = {'_frame': frame}         # Allow access to frame object.
    d.update(frame.f_globals)  # Unless shadowed by global
    d.update(frame.f_locals)

    i = code.InteractiveConsole(d)
    message = "Signal recieved : entering python shell.\nTraceback:\n"
    message += ''.join(traceback.format_stack(frame))
    i.interact(message)

def handle(self, conn, _address): # pylint: disable=method-hidden
        """
        Interact with one remote user.

        .. versionchanged:: 1.1b2 Each connection gets its own
            ``locals`` dictionary. Previously they were shared in a
            potentially unsafe manner.
        """
        fobj = conn.makefile(mode="rw")
        fobj = _fileobject(conn, fobj, self.stderr)
        getcurrent()._fileobj = fobj

        getcurrent().switch_in()
        try:
            console = InteractiveConsole(self._create_interactive_locals())
            if sys.version_info[:3] >= (3, 6, 0):
                # Beginning in 3.6, the console likes to print "now exiting <class>"
                # but probably our socket is already closed, so this just causes problems.
                console.interact(banner=self.banner, exitmsg='')
            else:
                console.interact(banner=self.banner)
        except SystemExit:  # raised by quit()
            if hasattr(sys, 'exc_clear'): # py2
                sys.exc_clear()
        finally:
            conn.close()
            fobj.close()

def __init__(self, items=None):
        if items is None:
            items = {}
        code.InteractiveConsole.__init__(self, items)
        self._buffer = ""

def main():
    url = sys.argv[1]
    context['url'] = url
    pkg = app.dispatch_url(url)
    context['pkg'] = pkg
    for item in pkg.to_dict().items():
        print '{} = {}'.format(*item)

    def prepare_readline():
        import os
        import readline
        import atexit

        readline.parse_and_bind('tab: complete')
        histfile = os.path.expanduser("~/.daenerys_history")

        try:
            readline.read_history_file(histfile)
        except IOError:
            pass

        def savehist(histfile):
            readline.write_history_file(histfile)

        atexit.register(savehist, histfile)
        del atexit

    try:
        from IPython.terminal.interactiveshell import TerminalInteractiveShell
        shell = TerminalInteractiveShell(user_ns=context)
        shell.mainloop()
    except ImportError:
        import code
        shell = code.InteractiveConsole(locals=context)
        shell.runcode(prepare_readline.__code__)
        shell.interact()

def debug():
    """Enter debug mode"""
    import readline  # optional, will allow Up/Down/History in the console
    import code
    def console_exit():
        raise SystemExit
    vars_copy = globals().copy()
    vars_copy.update(locals())
    vars_copy['exit'] = console_exit
    banner = 'Entering interactive debug mode\nType exit() to stop'
    shell = code.InteractiveConsole(vars_copy)
    try:
        shell.interact(banner=banner)
    except SystemExit:
        pass

def debug(sig, frame):
    """Interrupt running process, and provide a python prompt for
    interactive debugging."""
    d = {'_frame': frame}         # Allow access to frame object.
    d.update(frame.f_globals)  # Unless shadowed by global
    d.update(frame.f_locals)

    i = code.InteractiveConsole(d)
    message = "Signal received : entering python shell.\nTraceback:\n"
    message += ''.join(traceback.format_stack(frame))
    i.interact(message)

def embed():
      vars = globals()
      vars.update(locals())
      shell = code.InteractiveConsole(vars)
      shell.interact()